diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a4515fd2..5b804aadd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,52 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.20.0] - 2026-08-13 12:00:00 + +### Added + +- Addresses Issue [#1205](https://github.com/PSLmodels/OG-Core/issues/1205): + the UN Data Portal API token can now come from a `un_token` argument to + `demographics.get_un_data`, from a `UN_API_TOKEN` environment variable, or + from a single per-user file (`$XDG_CONFIG_HOME/og/un_api_token.txt`, or + `%APPDATA%\og\un_api_token.txt` on Windows). Sources are tried in that + order, then `un_api_token.txt` in the working directory. This gives one + token per user instead of one per directory. +- An `og-token` command, OG-Core's first console script, to manage that + token without hunting for the file: `og-token set` saves one, + `og-token show` reports where it lives and which source wins, and + `og-token rm` deletes it. The token is read without echoing and is never + printed back. + +### Changed + +- Answering the UN API token prompt now saves the token to the per-user file + rather than to the current working directory, which used to leave a copy of + the token in every directory a model was run from. An existing + `un_api_token.txt` in the working directory is still read, with a notice + that the location is deprecated. +- The token prompt now names both ways forward: where to generate a free + token, and that pressing return uses the archived copy of the same data + instead. +- The token is read with `getpass` rather than `input`, so it is no longer + echoed into the terminal and its scrollback while being typed or pasted. +- An expired token is now named as the reason for falling back to the + archived data, instead of failing the same way as a missing one. The + portal issues JSON Web Tokens, so the expiry date is read locally with + no extra request, and `og-token show` reports whether the stored token + is still current. A token that is not a readable JSON Web Token is used + as before, with no expiry reported. +- A run that falls back to the archived data now says once, not once per + series, how to register a token: where to get one, the full path of the + `og-token` command belonging to the running interpreter, and the token + file to write. The full path matters because `og-token` is installed + beside the interpreter and is normally not on the shell's PATH, so + someone who obtains a token days later has something they can paste + rather than a command that reports "not found". +- The token prompt is skipped when standard input is not interactive, so + scheduled and scripted runs fall back to the Population-Data archive + instead of waiting on input. + ## [0.19.2] - 2026-08-18 17:00:00 ### Adds @@ -674,6 +720,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Any earlier versions of OG-USA can be found in the [`OG-Core`](https://github.com/PSLmodels/OG-Core) repository [release history](https://github.com/PSLmodels/OG-Core/releases) from [v.0.6.4](https://github.com/PSLmodels/OG-Core/releases/tag/v0.6.4) (Jul. 20, 2021) or earlier. +[0.20.0]: https://github.com/PSLmodels/OG-Core/compare/v0.19.2...v0.20.0 [0.19.2]: https://github.com/PSLmodels/OG-Core/compare/v0.19.1...v0.19.2 [0.19.1]: https://github.com/PSLmodels/OG-Core/compare/v0.19.0...v0.19.1 [0.19.0]: https://github.com/PSLmodels/OG-Core/compare/v0.18.1...v0.19.0 diff --git a/docs/book/content/api/demographics.rst b/docs/book/content/api/demographics.rst index 15dc1a1c7..da3a47801 100644 --- a/docs/book/content/api/demographics.rst +++ b/docs/book/content/api/demographics.rst @@ -9,5 +9,7 @@ ogcore.demographics ------------------------------------------ .. automodule:: ogcore.demographics - :members: get_un_data, get_fert, get_mort, get_pop, pop_rebin, get_imm_rates, - immsolve, expand_pop_obj_J, get_pop_objs + :members: un_token_path, og_token_command, un_token_expiry, + resolve_un_token, get_un_data, get_fert, get_mort, get_pop, + pop_rebin, get_imm_rates, immsolve, expand_pop_obj_J, + get_pop_objs diff --git a/ogcore/__init__.py b/ogcore/__init__.py index 6ef897356..a77e3afbb 100644 --- a/ogcore/__init__.py +++ b/ogcore/__init__.py @@ -21,4 +21,4 @@ from ogcore.txfunc import * # noqa: F403 from ogcore.utils import * # noqa: F403 -__version__ = "0.19.2" +__version__ = "0.20.0" diff --git a/ogcore/demographics.py b/ogcore/demographics.py index 15114811f..2108b64b4 100644 --- a/ogcore/demographics.py +++ b/ogcore/demographics.py @@ -8,6 +8,12 @@ # Import packages import os +import sys +import argparse +import base64 +import datetime +import getpass +import json import numpy as np from io import StringIO import scipy.optimize as opt @@ -18,6 +24,15 @@ START_YEAR = 2024 END_YEAR = 2024 UN_COUNTRY_CODE = "840" # UN code for USA +UN_TOKEN_FILENAME = "un_api_token.txt" +UN_TOKEN_URL = "https://population.un.org/dataportalapi/index.html" +UN_DATA_ARCHIVE_URL = "https://github.com/EAPD-DRB/Population-Data" +# Warn only once per session about a token found in the working directory +_WARNED_LEGACY_UN_TOKEN = False +# Say how to register a token only once per session, not on every request +_HINTED_NO_UN_TOKEN = False +# Say a token has expired only once per session +_WARNED_EXPIRED_UN_TOKEN = False # create output director for figures CUR_PATH = os.path.split(os.path.abspath(__file__))[0] OUTPUT_DIR = os.path.join(CUR_PATH, "..", "data", "OUTPUT", "Demographics") @@ -32,11 +47,379 @@ """ +def un_token_path(): + """ + This function returns the path of the per-user file that holds the UN + Data Portal API token. The location follows the platform convention + for user configuration files: ``$XDG_CONFIG_HOME`` (or ``~/.config`` + when that is unset) on macOS and Linux, and ``%APPDATA%`` on Windows. + + Returns: + path (str): full path to the user's UN API token file + """ + if os.name == "nt": + base = os.environ.get("APPDATA") or os.path.expanduser("~") + else: + base = os.environ.get("XDG_CONFIG_HOME") or os.path.join( + os.path.expanduser("~"), ".config" + ) + + return os.path.join(base, "og", UN_TOKEN_FILENAME) + + +def _clean_un_token(un_token): + """ + This function normalizes a UN Data Portal API token by removing + surrounding whitespace and any leading "Bearer " prefix, so that the + request header is not doubled into "Bearer Bearer ". + + Args: + un_token (str): raw token, may be None + + Returns: + un_token (str): normalized token, empty string if none was given + """ + un_token = (un_token or "").strip() + if un_token.lower().startswith("bearer "): + un_token = un_token[len("bearer ") :].strip() + + return un_token + + +def un_token_expiry(un_token): + """ + This function reads the expiry date out of a UN Data Portal API token. + The portal issues JSON Web Tokens, whose middle segment carries an + ``exp`` claim, so the date can be read without a network call. The + signature is not checked and is not needed here: the portal remains + the authority on whether a token is accepted, and this is only used to + tell a user that renewing is due. + + Args: + un_token (str): token to inspect + + Returns: + expiry (datetime.date): expiry date, or None when the token is not + a readable JSON Web Token + """ + try: + payload = un_token.split(".")[1] + payload += "=" * (-len(payload) % 4) + claims = json.loads(base64.urlsafe_b64decode(payload)) + + return datetime.datetime.fromtimestamp( + claims["exp"], datetime.timezone.utc + ).date() + except (AttributeError, IndexError, KeyError, TypeError, ValueError): + return None # opaque token, or a format we do not recognize + + +def _utc_today(): + """Today's date in UTC, to compare against a token's expiry claim.""" + return datetime.datetime.now(datetime.timezone.utc).date() + + +def _warn_if_un_token_expired(un_token): + """ + This function says, once per session, that a token has expired. An + expired token otherwise fails in the same silent-looking way as a + missing one: the request is refused and the caller quietly falls back + to the archived data. + + Args: + un_token (str): the token that was resolved + + Returns: + None + """ + global _WARNED_EXPIRED_UN_TOKEN + if _WARNED_EXPIRED_UN_TOKEN: + return + expiry = un_token_expiry(un_token) + if expiry is None or expiry >= _utc_today(): + return + _WARNED_EXPIRED_UN_TOKEN = True + + lines = [ + f"Your UN API token expired on {expiry}, so the archived data " + "will be used.", + f" Get a new one at {UN_TOKEN_URL}", + ] + command = og_token_command() + if command: + lines.append(f" Then run: {command} set") + else: + lines.append(f" Then save it to {un_token_path()}") + print("\n".join(lines)) + + +def og_token_command(): + """ + This function returns the full path of the ``og-token`` command that + belongs to the running interpreter, so that a message can tell the + user exactly what to type. The command is installed beside the + interpreter and is usually not on the shell's PATH, because OG-Core is + normally run from a project virtual environment. + + Returns: + command (str): full path to og-token, or None when it is not + installed alongside this interpreter + """ + name = "og-token.exe" if os.name == "nt" else "og-token" + command = os.path.join(os.path.dirname(sys.executable), name) + + return command if os.path.exists(command) else None + + +def _hint_how_to_register_token(): + """ + This function prints, once per session, how to register a token. It + runs whenever a request falls back to the archived data, so a user who + obtains a token later is told what to do without being prompted again + on every run. + """ + global _HINTED_NO_UN_TOKEN + if _HINTED_NO_UN_TOKEN: + return + _HINTED_NO_UN_TOKEN = True + + lines = [ + "No UN API token registered, so the archived data will be used.", + f" Get a free token at {UN_TOKEN_URL}", + ] + command = og_token_command() + if command: + lines.append(f" Then run: {command} set") + lines.append(f" (or save the token to {un_token_path()})") + else: + lines.append(f" Then save the token to {un_token_path()}") + print("\n".join(lines)) + + +def resolve_un_token(un_token=None): + """ + This function finds the UN Data Portal API token to use for a + request. Sources are tried in order and the first one that is present + wins: + + 1. the ``un_token`` argument + 2. the ``UN_API_TOKEN`` environment variable + 3. the per-user file at :func:`un_token_path` + 4. ``un_api_token.txt`` in the current working directory (deprecated) + + When no source holds a token the user is asked for one and the answer + is saved to the per-user file, so a token is entered once per machine + rather than once per directory. The prompt is skipped when standard + input is not interactive, in which case an empty token is returned and + the caller falls back to the Population-Data archive. + + Args: + un_token (str): token supplied by the caller, overrides all other + sources + + Returns: + un_token (str): normalized token, empty string if none was found + """ + un_token = _find_un_token(un_token) + if un_token: + _warn_if_un_token_expired(un_token) + else: + _hint_how_to_register_token() + + return un_token + + +def _find_un_token(un_token=None): + """ + This function does the source-by-source lookup described in + :func:`resolve_un_token`, which wraps it to add the one-time hint when + nothing is found. + + Args: + un_token (str): token supplied by the caller + + Returns: + un_token (str): normalized token, empty string if none was found + """ + global _WARNED_LEGACY_UN_TOKEN + + if un_token: + return _clean_un_token(un_token) + + # .strip() so a variable set to blank space falls through to the files + # rather than silently resolving to no token at all. + if os.environ.get("UN_API_TOKEN", "").strip(): + return _clean_un_token(os.environ["UN_API_TOKEN"]) + + # An existing per-user file is authoritative even when empty, so that + # a user who declined the prompt is not asked again on every call. + user_path = un_token_path() + if os.path.exists(user_path): + with open(user_path, "r") as file: + return _clean_un_token(file.read()) + + if os.path.exists(UN_TOKEN_FILENAME): + if not _WARNED_LEGACY_UN_TOKEN: + print( + f"Using the UN API token in {UN_TOKEN_FILENAME} in the " + "current directory. This location is deprecated because it " + "leaves a copy of the token in every directory you run " + f"from. Move it to {user_path} to keep one token per user." + ) + _WARNED_LEGACY_UN_TOKEN = True + with open(UN_TOKEN_FILENAME, "r") as file: + return _clean_un_token(file.read()) + + try: + if not sys.stdin or not sys.stdin.isatty(): + return "" # not interactive, e.g. a scheduled run + print( + "\nOG-Core can read population data directly from the UN Data " + "Portal, which needs a free API token.\n" + f" To get one, open {UN_TOKEN_URL} and click Generate Token.\n" + " Or press return to use the archived copy of the same data " + f"at {UN_DATA_ARCHIVE_URL}.\n" + ) + # getpass rather than input so the token is not echoed into the + # terminal and its scrollback. + un_token = getpass.getpass("UN API token (input is hidden): ") + except (EOFError, ValueError): # stdin at end of file or closed + return "" + + # Save the answer, empty or not, so the question is asked only once. + try: + os.makedirs(os.path.dirname(user_path), exist_ok=True) + with open(user_path, "w") as file: + file.write(un_token) + except OSError as err: # e.g. a read-only home directory + print( + f"Could not save the UN API token to {user_path} ({err}). " + "It will be used for this session only." + ) + else: + try: + os.chmod(user_path, 0o600) + except OSError: # permissions are not settable on every platform + pass + if _clean_un_token(un_token): + print(f"Token saved to {user_path}") + + return _clean_un_token(un_token) + + +def un_token_cli(argv=None): + """ + This function is the command line entry point for managing the stored + UN Data Portal API token. It is installed as ``og-token`` and takes one + of three actions: ``set`` saves a token to the per-user file, ``show`` + reports where the token lives and which source would be used, and + ``rm`` deletes the stored token. + + Args: + argv (list): command line arguments, read from sys.argv when not + given + + Returns: + status (int): process exit status, 0 on success + """ + parser = argparse.ArgumentParser( + prog="og-token", + description=( + "Manage the UN Data Portal API token used by OG-Core. Get a " + f"free token from {UN_TOKEN_URL} (click Generate Token). " + "Without one, OG-Core reads the archived copy of the same data " + f"from {UN_DATA_ARCHIVE_URL}." + ), + ) + parser.add_argument( + "action", + choices=["set", "show", "rm"], + help="save a token, report where it lives, or delete it", + ) + args = parser.parse_args(argv) + path = un_token_path() + + if args.action == "set": + print(f"Get a free token at {UN_TOKEN_URL} (click Generate Token).") + try: + un_token = _clean_un_token(getpass.getpass("UN API token: ")) + except (EOFError, KeyboardInterrupt): + print("\nCancelled. Nothing was saved.") + return 1 + if not un_token: + print("No token entered. Nothing was saved.") + return 1 + try: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as file: + file.write(un_token) + except OSError as err: + print(f"Could not write {path} ({err}).") + return 1 + try: + os.chmod(path, 0o600) + except OSError: # permissions are not settable on every platform + pass + expiry = un_token_expiry(un_token) + if expiry is None: + print(f"Token saved to {path}") + elif expiry < _utc_today(): + print(f"Token saved to {path}, but it expired on {expiry}.") + print(f"Get a current one at {UN_TOKEN_URL}") + else: + print(f"Token saved to {path}, valid until {expiry}") + return 0 + + if args.action == "rm": + if not os.path.exists(path): + print(f"No token stored at {path}") + return 1 + try: + os.remove(path) + except OSError as err: + print(f"Could not remove {path} ({err}).") + return 1 + print(f"Removed {path}") + return 0 + + # show + print(f"Token file: {path}") + if os.path.exists(path): + with open(path, "r") as file: + stored = _clean_un_token(file.read()) + if not stored: + print(" present but empty, so no token is sent") + else: + expiry = un_token_expiry(stored) + if expiry is None: + print(" a token is stored") + elif expiry < _utc_today(): + print(f" a token is stored, but it expired on {expiry}") + print(f" get a new one at {UN_TOKEN_URL}") + else: + days = (expiry - _utc_today()).days + print( + f" a token is stored, valid until {expiry} ({days} days)" + ) + else: + print(" not set, run 'og-token set'") + if os.environ.get("UN_API_TOKEN", "").strip(): + print("UN_API_TOKEN is set and takes precedence over the file.") + if os.path.exists(UN_TOKEN_FILENAME): + print( + f"A deprecated {UN_TOKEN_FILENAME} is in this directory. It is " + "only used when neither of the above is set." + ) + + return 0 + + def get_un_data( variable_code, country_id=UN_COUNTRY_CODE, start_year=START_YEAR, end_year=END_YEAR, + un_token=None, ): """ This function retrieves data from the United Nations Data Portal API @@ -48,6 +431,8 @@ def get_un_data( country_id (str): country id for UN data start_year (int): start year for UN data end_year (int): end year for UN data + un_token (str): UN Data Portal API token, resolved from the + environment or the user's token file when not given Returns: df (Pandas DataFrame): DataFrame of UN data @@ -64,30 +449,9 @@ def get_un_data( + "?format=csv" ) - # Check for a file named "un_api_token.txt" in the current directory - if os.path.exists(os.path.join("un_api_token.txt")): - with open(os.path.join("un_api_token.txt"), "r") as file: - UN_TOKEN = file.read().strip() - else: # if file not exist, prompt user for token - try: - UN_TOKEN = input( - "Please enter your UN API token " - "(press return if you do not have one): " - ) - # write the UN_TOKEN to a file to find in the future - with open(os.path.join("un_api_token.txt"), "w") as file: - file.write(UN_TOKEN) - except EOFError: - UN_TOKEN = "" - # get data from url payload = {} - # Accept a token with or without a leading "Bearer " prefix so the - # header isn't doubled into "Bearer Bearer ". - UN_TOKEN = UN_TOKEN.strip() - if UN_TOKEN.lower().startswith("bearer "): - UN_TOKEN = UN_TOKEN[len("bearer ") :].strip() - headers = {"Authorization": "Bearer " + UN_TOKEN} + headers = {"Authorization": "Bearer " + resolve_un_token(un_token)} response = get_legacy_session().get(target, headers=headers, data=payload) # Check if the request was successful before processing if response.status_code == 200: diff --git a/pyproject.toml b/pyproject.toml index 089735c8f..64fb89c69 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "ogcore" -version = "0.19.2" +version = "0.20.0" authors = [ {name = "Jason DeBacker and Richard W. Evans"}, ] @@ -41,6 +41,9 @@ dependencies = [ Homepage = "https://github.com/PSLmodels/OG-Core/" "Issue Tracker" = "https://github.com/PSLmodels/OG-Core/issues" +[project.scripts] +og-token = "ogcore.demographics:un_token_cli" + [project.optional-dependencies] dev = [ "pytest>=9.1", diff --git a/tests/test_demographics.py b/tests/test_demographics.py index 14fe5fbf6..32673ec5d 100644 --- a/tests/test_demographics.py +++ b/tests/test_demographics.py @@ -1,6 +1,9 @@ import numpy as np import pytest import os +import base64 +import datetime +import json from ogcore import demographics # Read in some test population data to use in select tests below @@ -820,3 +823,427 @@ def test_expand_pop_obj_J_preserves_aggregate_mortality_with_gradients(): assert np.allclose(omega.sum(axis=2), omega_path_S) assert np.allclose((within_age_weights * rho).sum(axis=2), mort_rates_S) assert np.all((rho >= 0) & (rho <= 1)) + + +""" +------------------------------------------------------------------------ +Tests of the UN Data Portal API token resolution +------------------------------------------------------------------------ +""" + + +@pytest.fixture +def isolated_token_env(monkeypatch, tmp_path): + """ + Point token resolution at a temporary home and working directory so + the tests never read or write the developer's real token. + """ + home = tmp_path / "home" + cwd = tmp_path / "cwd" + home.mkdir() + cwd.mkdir() + monkeypatch.delenv("UN_API_TOKEN", raising=False) + monkeypatch.setenv("XDG_CONFIG_HOME", str(home / ".config")) + monkeypatch.setenv("APPDATA", str(home / "AppData")) + monkeypatch.setattr(demographics, "_WARNED_LEGACY_UN_TOKEN", False) + monkeypatch.setattr(demographics, "_HINTED_NO_UN_TOKEN", False) + monkeypatch.chdir(cwd) + return cwd + + +class _Stdin: + """Stand-in for sys.stdin with a controllable isatty().""" + + def __init__(self, interactive): + self.interactive = interactive + + def isatty(self): + return self.interactive + + +def _set_tty(monkeypatch, interactive): + """ + Control whether resolution believes the session is interactive. + pytest captures stdin, so isatty() is False by default and the prompt + path has to be switched on explicitly. + """ + monkeypatch.setattr("sys.stdin", _Stdin(interactive)) + + +def test_un_token_path_follows_platform_convention(monkeypatch, tmp_path): + """The token file sits under the user's config directory.""" + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "cfg")) + monkeypatch.setenv("APPDATA", str(tmp_path / "cfg")) + path = demographics.un_token_path() + assert path.startswith(str(tmp_path / "cfg")) + assert path.endswith(demographics.UN_TOKEN_FILENAME) + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("abc", "abc"), + (" abc ", "abc"), + ("Bearer abc", "abc"), + ("bearer abc ", "abc"), + (None, ""), + ("", ""), + ], + ids=["plain", "padded", "bearer", "lower_bearer", "none", "empty"], +) +def test_clean_un_token(raw, expected): + """Whitespace and any 'Bearer ' prefix are removed.""" + assert demographics._clean_un_token(raw) == expected + + +def test_argument_wins_over_every_other_source( + isolated_token_env, monkeypatch +): + """An explicit token beats the environment and both files.""" + monkeypatch.setenv("UN_API_TOKEN", "from_env") + (isolated_token_env / demographics.UN_TOKEN_FILENAME).write_text( + "from_cwd" + ) + assert demographics.resolve_un_token("from_arg") == "from_arg" + + +def test_env_var_wins_over_files(isolated_token_env, monkeypatch): + """The environment variable beats both the user file and the cwd file.""" + monkeypatch.setenv("UN_API_TOKEN", "Bearer from_env") + user_path = demographics.un_token_path() + os.makedirs(os.path.dirname(user_path), exist_ok=True) + with open(user_path, "w") as f: + f.write("from_user_file") + (isolated_token_env / demographics.UN_TOKEN_FILENAME).write_text( + "from_cwd" + ) + assert demographics.resolve_un_token() == "from_env" + + +def test_user_file_wins_over_cwd_file(isolated_token_env): + """The per-user file beats the deprecated working-directory file.""" + user_path = demographics.un_token_path() + os.makedirs(os.path.dirname(user_path), exist_ok=True) + with open(user_path, "w") as f: + f.write("from_user_file\n") + (isolated_token_env / demographics.UN_TOKEN_FILENAME).write_text( + "from_cwd" + ) + assert demographics.resolve_un_token() == "from_user_file" + + +def test_empty_user_file_is_authoritative(isolated_token_env, monkeypatch): + """ + A user who declined the prompt is not asked again, and the cwd file is + not consulted behind their back. + """ + user_path = demographics.un_token_path() + os.makedirs(os.path.dirname(user_path), exist_ok=True) + with open(user_path, "w") as f: + f.write("") + (isolated_token_env / demographics.UN_TOKEN_FILENAME).write_text( + "from_cwd" + ) + + def _fail(*args, **kwargs): + raise AssertionError("the user should not be prompted again") + + monkeypatch.setattr(demographics.getpass, "getpass", _fail) + assert demographics.resolve_un_token() == "" + + +def test_cwd_file_still_works_and_warns(isolated_token_env, capsys): + """The old location keeps working, with a one-time deprecation notice.""" + (isolated_token_env / demographics.UN_TOKEN_FILENAME).write_text( + "from_cwd" + ) + assert demographics.resolve_un_token() == "from_cwd" + assert "deprecated" in capsys.readouterr().out + # the notice is not repeated on later calls in the same session + assert demographics.resolve_un_token() == "from_cwd" + assert "deprecated" not in capsys.readouterr().out + + +def test_prompt_saves_to_the_user_file_not_the_working_directory( + isolated_token_env, monkeypatch +): + """This is the behavior change: answering the prompt no longer leaves a + copy of the token in whatever directory the run started from.""" + _set_tty(monkeypatch, True) + monkeypatch.setattr( + demographics.getpass, "getpass", lambda *a, **k: "typed_token" + ) + + assert demographics.resolve_un_token() == "typed_token" + + user_path = demographics.un_token_path() + assert os.path.exists(user_path) + with open(user_path) as f: + assert f.read() == "typed_token" + assert not os.path.exists( + isolated_token_env / demographics.UN_TOKEN_FILENAME + ) + + +def test_no_prompt_when_not_interactive(isolated_token_env, monkeypatch): + """A scheduled run gets an empty token instead of hanging, and writes + nothing to disk.""" + _set_tty(monkeypatch, False) + + def _fail(*args, **kwargs): + raise AssertionError("a non-interactive session must not prompt") + + monkeypatch.setattr(demographics.getpass, "getpass", _fail) + assert demographics.resolve_un_token() == "" + assert not os.path.exists(demographics.un_token_path()) + + +def test_get_un_data_sends_the_resolved_token(isolated_token_env, monkeypatch): + """The token reaches the request header, so the wiring from argument to + Authorization is covered and not just the resolver in isolation.""" + sent = {} + + class _Response: + status_code = 401 # short-circuits before any parsing + + class _Session: + def get(self, target, headers=None, data=None): + sent["headers"] = headers + return _Response() + + monkeypatch.setattr(demographics, "get_legacy_session", lambda: _Session()) + demographics.get_un_data("47", un_token="Bearer explicit_token") + assert sent["headers"]["Authorization"] == "Bearer explicit_token" + + monkeypatch.setenv("UN_API_TOKEN", "env_token") + demographics.get_un_data("47") + assert sent["headers"]["Authorization"] == "Bearer env_token" + + +def test_un_token_cli_set_show_rm(isolated_token_env, monkeypatch, capsys): + """The og-token command round-trips: save, report, delete.""" + monkeypatch.setattr( + demographics.getpass, "getpass", lambda *a, **k: "Bearer cli_tok1234" + ) + + assert demographics.un_token_cli(["set"]) == 0 + path = demographics.un_token_path() + with open(path) as f: + assert f.read() == "cli_tok1234" # the Bearer prefix is stripped + assert demographics.resolve_un_token() == "cli_tok1234" + + assert demographics.un_token_cli(["show"]) == 0 + out = capsys.readouterr().out + assert path in out + assert "a token is stored" in out + assert "cli_tok1234" not in out # no part of the token is printed + assert "1234" not in out + + assert demographics.un_token_cli(["rm"]) == 0 + assert not os.path.exists(path) + assert demographics.un_token_cli(["rm"]) == 1 # nothing left to remove + + +def test_un_token_cli_set_rejects_an_empty_answer( + isolated_token_env, monkeypatch, capsys +): + """Pressing return at the prompt writes nothing and reports failure.""" + monkeypatch.setattr(demographics.getpass, "getpass", lambda *a, **k: " ") + + assert demographics.un_token_cli(["set"]) == 1 + assert not os.path.exists(demographics.un_token_path()) + assert "Nothing was saved" in capsys.readouterr().out + + +def test_un_token_cli_show_flags_the_environment_override( + isolated_token_env, monkeypatch, capsys +): + """show warns when the environment variable will win over the file.""" + monkeypatch.setenv("UN_API_TOKEN", "env_token") + assert demographics.un_token_cli(["show"]) == 0 + assert "takes precedence" in capsys.readouterr().out + + +@pytest.mark.parametrize( + "interrupt", [EOFError, KeyboardInterrupt], ids=["eof", "ctrl_c"] +) +def test_un_token_cli_set_handles_an_interrupted_prompt( + isolated_token_env, monkeypatch, capsys, interrupt +): + """`og-token set < /dev/null` or Ctrl-C reports cleanly instead of + printing a traceback.""" + + def _raise(*args, **kwargs): + raise interrupt() + + monkeypatch.setattr(demographics.getpass, "getpass", _raise) + assert demographics.un_token_cli(["set"]) == 1 + assert "Cancelled" in capsys.readouterr().out + assert not os.path.exists(demographics.un_token_path()) + + +def test_prompt_offers_both_ways_out(isolated_token_env, monkeypatch, capsys): + """The prompt tells the user where to get a token and what happens if + they decline, rather than leaving them to guess.""" + _set_tty(monkeypatch, True) + monkeypatch.setattr(demographics.getpass, "getpass", lambda *a, **k: "") + + assert demographics.resolve_un_token() == "" + out = capsys.readouterr().out + assert demographics.UN_TOKEN_URL in out + assert demographics.UN_DATA_ARCHIVE_URL in out + + +def test_hint_names_a_runnable_command_and_the_file( + isolated_token_env, monkeypatch, capsys +): + """Falling back to the archive tells the user how to register a token. + `og-token` is installed beside the interpreter and is normally not on + the shell's PATH, so the hint has to give the full path, plus the file + as a way that needs no environment at all.""" + _set_tty(monkeypatch, False) # no prompt, straight to the fallback + + assert demographics.resolve_un_token() == "" + out = capsys.readouterr().out + assert demographics.UN_TOKEN_URL in out + assert demographics.un_token_path() in out # always actionable + + command = demographics.og_token_command() + if command is not None: + assert command in out + assert os.path.isabs(command) # copy-pasteable from any shell + + +def test_hint_is_printed_once_per_session( + isolated_token_env, monkeypatch, capsys +): + """get_pop_objs resolves a token once per series, so the hint must not + repeat three times in a single run.""" + _set_tty(monkeypatch, False) + + demographics.resolve_un_token() + first = capsys.readouterr().out + assert "No UN API token registered" in first + + demographics.resolve_un_token() + demographics.resolve_un_token() + assert "No UN API token registered" not in capsys.readouterr().out + + +def test_no_hint_when_a_token_is_present( + isolated_token_env, monkeypatch, capsys +): + """Users who have a token are not nagged.""" + monkeypatch.setenv("UN_API_TOKEN", "a_real_token") + assert demographics.resolve_un_token() == "a_real_token" + assert "No UN API token registered" not in capsys.readouterr().out + + +def test_token_is_never_echoed_at_the_prompt( + isolated_token_env, monkeypatch, capsys +): + """The token is a secret, so it is read with getpass and never reaches + the terminal or its scrollback. Reverting to input() would echo it.""" + _set_tty(monkeypatch, True) + + def _must_not_be_used(*args, **kwargs): + raise AssertionError("input() echoes; the token must use getpass") + + monkeypatch.setattr("builtins.input", _must_not_be_used) + monkeypatch.setattr( + demographics.getpass, "getpass", lambda *a, **k: "s3cret_token" + ) + + assert demographics.resolve_un_token() == "s3cret_token" + out = capsys.readouterr().out + assert "s3cret_token" not in out # not printed back either + + +def _make_jwt(days_from_today): + """Build a token shaped like the portal's: three base64url segments + with an `exp` claim. Only the payload matters here; the signature is + never checked.""" + exp = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta( + days=days_from_today + ) + + def seg(obj): + raw = json.dumps(obj).encode() + return base64.urlsafe_b64encode(raw).decode().rstrip("=") + + return ".".join( + [ + seg({"alg": "HS256", "typ": "JWT"}), + seg({"exp": int(exp.timestamp()), "unique_name": "someone"}), + "not_a_real_signature", + ] + ) + + +def test_un_token_expiry_reads_the_claim(): + """The portal issues JWTs, so the expiry is readable without a call.""" + expiry = demographics.un_token_expiry(_make_jwt(30)) + assert ( + expiry + == ( + datetime.datetime.now(datetime.timezone.utc) + + datetime.timedelta(30) + ).date() + ) + + +@pytest.mark.parametrize( + "value", + ["opaque-token-with-no-dots", "a.b.c", "", None, 12345, "a..c"], + ids=["opaque", "not_base64", "empty", "none", "not_a_string", "empty_seg"], +) +def test_un_token_expiry_degrades_quietly(value): + """Anything not a readable JWT returns None rather than raising, so a + model run never dies because the portal changed its token format.""" + assert demographics.un_token_expiry(value) is None + + +def test_expired_token_is_reported_once( + isolated_token_env, monkeypatch, capsys +): + """An expired token otherwise fails the same silent-looking way as a + missing one.""" + monkeypatch.setattr(demographics, "_WARNED_EXPIRED_UN_TOKEN", False) + expired = _make_jwt(-5) + monkeypatch.setenv("UN_API_TOKEN", expired) + + assert demographics.resolve_un_token() == expired # still returned + out = capsys.readouterr().out + assert "expired on" in out + assert demographics.UN_TOKEN_URL in out + assert expired not in out # the token itself is never printed + + demographics.resolve_un_token() + assert "expired on" not in capsys.readouterr().out # once per session + + +def test_valid_token_is_not_flagged(isolated_token_env, monkeypatch, capsys): + """A token with time left produces no noise.""" + monkeypatch.setattr(demographics, "_WARNED_EXPIRED_UN_TOKEN", False) + monkeypatch.setenv("UN_API_TOKEN", _make_jwt(60)) + + demographics.resolve_un_token() + assert "expired" not in capsys.readouterr().out + + +@pytest.mark.parametrize( + "days,expected", [(-5, "expired on"), (60, "valid until")] +) +def test_cli_show_reports_expiry( + isolated_token_env, monkeypatch, capsys, days, expected +): + """og-token show says whether the stored token is still usable.""" + token = _make_jwt(days) + monkeypatch.setattr(demographics.getpass, "getpass", lambda *a, **k: token) + assert demographics.un_token_cli(["set"]) == 0 + capsys.readouterr() + + assert demographics.un_token_cli(["show"]) == 0 + out = capsys.readouterr().out + assert expected in out + assert token not in out diff --git a/uv.lock b/uv.lock index 3dbd3fcc9..4975e4a5b 100644 --- a/uv.lock +++ b/uv.lock @@ -1586,7 +1586,7 @@ wheels = [ [[package]] name = "ogcore" -version = "0.19.2" +version = "0.20.0" source = { editable = "." } dependencies = [ { name = "dask" },