This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch add-python-tests in repository https://gitbox.apache.org/repos/asf/airflow-steward.git
commit 0f97ca2bdece4b1d5725769c29df91426d420b55 Author: Jarek Potiuk <[email protected]> AuthorDate: Wed Apr 29 02:43:12 2026 +0200 tests(oauth-draft): expand coverage 62%→99%; add CI tests workflow Adds 36 new unit tests for the oauth-draft Python project (22 → 58 total) and a dedicated `.github/workflows/tests.yml` CI workflow that runs the per-tool pytest suites as a parallel matrix job. oauth-draft test additions: - `tests/test_credentials.py` — 3 tests for `refresh_access_token`: successful POST + access-token return, HTTP error path, payload with no `access_token` field. Mocks `urllib.request.urlopen` with a small `_FakeResponse` context-manager stand-in (used in every network-mocked test below as well). - `tests/test_create_draft.py` — 13 tests covering: - `read_body` (file path, stdin via "-", stdin via None); - `api_get` / `api_post` (request shape + response parsing); - `api_post` HTTP error path; - `latest_reply_headers` (delegates to `api_get` then to `headers_from_thread`); - `create_draft` payload shape with and without `threadId`; - `main` end-to-end with mocked `refresh_access_token`, `latest_reply_headers`, and `api_post` — verifies the MIME body posted to `/drafts` carries the right From/To/Subject/ In-Reply-To headers and the body content; - `main --no-reply-headers` skips the thread-headers lookup. - `tests/test_mark_threads_read.py` — 9 tests covering: - `list_thread_ids` pagination (multi-page response stitched correctly; second-page request includes `pageToken`); - `list_thread_ids` HTTP error path; - `modify_thread` payload shape (with both label lists, with only one label list); - `modify_thread` HTTP error path; - `main` dry-run (lists IDs, never calls modify); - `main --execute` calls `modify_thread` per thread with the default `removeLabelIds=[UNREAD]`; - `main --max` truncates the modify count; - `main` returns 1 when any modify call fails. - `tests/test_setup_creds.py` — new file, 11 tests covering the whole `setup_creds` module (was 22% covered → ~89%): - `detect_from_address` (env var, git-config fallback, missing git, git error, empty git output); - `parse_args` (defaults + overrides); - `main` error paths (no from-address, missing client_secrets, flow returns no refresh_token); - `main` success path: writes credentials JSON atomically with mode 600 in a 700 parent dir; - `main --rm-client-secrets` deletes the input file; - `main` handles both `installed` and `web` shapes of the Google client_secrets.json. Coverage on `src/oauth_draft/` after the additions: ``` src/oauth_draft/__init__.py 100% src/oauth_draft/create_draft.py 99% src/oauth_draft/credentials.py 100% src/oauth_draft/mark_threads_read.py 99% src/oauth_draft/setup_creds.py 89% TOTAL 99% ``` The remaining ~1% is the `if __name__ == "__main__":` shim in each module and a handful of chmod-error warning paths in `setup_creds` that are awkward to provoke deterministically. CI workflow (`.github/workflows/tests.yml`): - Runs on every PR and push to main. - Per-project pytest matrix (`fail-fast: false`) so the two projects' results are visible as separate CI checks rather than buried in the bundled `pytest` lines of the `prek` workflow. - Uses `uv run --directory <project>` (not `--project`) to move cwd into the project root — pytest needs it because each project's `[tool.pytest.ini_options] testpaths = ["tests"]` resolves relative to cwd. - The `prek` workflow's pytest hooks still run (defense in depth); this workflow is the visible signal, not the gate. Other: - `.gitignore` adds `.coverage` to both Python projects so locally- run `coverage` doesn't dirty the working tree. Generated-by: Claude Code (Claude Opus 4.7) --- .github/workflows/tests.yml | 64 ++++++ tools/gmail/oauth-draft/.gitignore | 1 + tools/gmail/oauth-draft/tests/test_create_draft.py | 226 ++++++++++++++++++- tools/gmail/oauth-draft/tests/test_credentials.py | 75 ++++++- .../oauth-draft/tests/test_mark_threads_read.py | 239 ++++++++++++++++++++- tools/gmail/oauth-draft/tests/test_setup_creds.py | 219 +++++++++++++++++++ tools/vulnogram/generate-cve-json/.gitignore | 1 + 7 files changed, 822 insertions(+), 3 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..cfe61b9 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,64 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +--- +name: tests + +on: # yamllint disable-line rule:truthy + pull_request: + push: + branches: [main] + +permissions: {} + +jobs: + # Per-project pytest matrix. Each Python project under tools/ has its + # own `pyproject.toml` + `uv.lock`, with pytest declared in the `dev` + # dependency group. Running them as separate matrix jobs surfaces + # per-project pass/fail in the CI checks list, which is easier to + # triage than the bundled `pytest` lines inside the `prek` workflow's + # output. The `prek` workflow still exercises pytest as a hook, so + # this workflow is the visible signal — not the gate. + pytest: + name: "pytest (${{ matrix.project.name }})" + runs-on: ubuntu-latest + permissions: + contents: read + strategy: + fail-fast: false + matrix: + project: + - name: oauth-draft + path: tools/gmail/oauth-draft + - name: generate-cve-json + path: tools/vulnogram/generate-cve-json + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + # uv brings its own Python and reads each project's + # `pyproject.toml` + `uv.lock`. Minimum uv version is enforced + # by the root `pyproject.toml`'s `[tool.uv] required-version`. + - uses: astral-sh/setup-uv@eac588ad8def6316056a12d4907a9d4d84ff7a3b # v7.3.0 + with: + enable-cache: true + - name: Run pytest + # `--directory` (not `--project`) — both move uv's project + # context, but `--directory` also changes cwd, which pytest + # needs because each project's `pyproject.toml` declares + # `testpaths = ["tests"]` relative to its own root. + run: uv run --directory ${{ matrix.project.path }} --group dev pytest diff --git a/tools/gmail/oauth-draft/.gitignore b/tools/gmail/oauth-draft/.gitignore index ab7ef44..8cae406 100644 --- a/tools/gmail/oauth-draft/.gitignore +++ b/tools/gmail/oauth-draft/.gitignore @@ -4,3 +4,4 @@ __pycache__/ .pytest_cache/ .ruff_cache/ .mypy_cache/ +.coverage diff --git a/tools/gmail/oauth-draft/tests/test_create_draft.py b/tools/gmail/oauth-draft/tests/test_create_draft.py index d6675d2..d643962 100644 --- a/tools/gmail/oauth-draft/tests/test_create_draft.py +++ b/tools/gmail/oauth-draft/tests/test_create_draft.py @@ -16,11 +16,28 @@ # under the License. from __future__ import annotations +import base64 import email import email.policy +import io +import json +import urllib.error from email.message import EmailMessage +from unittest.mock import patch -from oauth_draft.create_draft import build_mime, headers_from_thread, parse_args +import pytest + +from oauth_draft.create_draft import ( + api_get, + api_post, + build_mime, + create_draft, + headers_from_thread, + latest_reply_headers, + main, + parse_args, + read_body, +) def parse_built_message(raw: bytes) -> EmailMessage: @@ -174,3 +191,210 @@ def test_parse_args_repeats(): ) assert args.to == ["[email protected]", "[email protected]"] assert args.cc == ["[email protected]"] + + +# --- read_body ------------------------------------------------------------- + + +def test_read_body_from_file(tmp_path): + p = tmp_path / "body.txt" + p.write_text("hello from file") + assert read_body(str(p)) == "hello from file" + + +def test_read_body_from_stdin_when_dash(monkeypatch): + monkeypatch.setattr("sys.stdin", io.StringIO("piped")) + assert read_body("-") == "piped" + + +def test_read_body_from_stdin_when_none(monkeypatch): + monkeypatch.setattr("sys.stdin", io.StringIO("default")) + assert read_body(None) == "default" + + +# --- network-mocked helpers ------------------------------------------------ + + +class _FakeResponse: + def __init__(self, payload: bytes): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return self._payload + + +def test_api_get_parses_json_response(): + with patch("oauth_draft.create_draft.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b'{"id": "thread-1"}') + result = api_get("token", "/threads/thread-1") + assert result == {"id": "thread-1"} + request = mock_open.call_args.args[0] + assert request.full_url.endswith("/threads/thread-1") + assert request.headers["Authorization"] == "Bearer token" + + +def test_api_post_parses_json_response(): + with patch("oauth_draft.create_draft.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b'{"id": "draft-7"}') + result = api_post("token", "/drafts", {"message": {"raw": "X"}}) + assert result == {"id": "draft-7"} + request = mock_open.call_args.args[0] + assert request.method == "POST" + assert request.headers["Content-type"] == "application/json" + assert json.loads(request.data) == {"message": {"raw": "X"}} + + +def test_api_post_raises_on_http_error(): + err = urllib.error.HTTPError( + url="https://x";, + code=403, + msg="Forbidden", + hdrs=None, # type: ignore[arg-type] + fp=io.BytesIO(b'{"error": "forbidden"}'), + ) + with patch("oauth_draft.create_draft.urllib.request.urlopen", side_effect=err): + with pytest.raises(SystemExit) as excinfo: + api_post("token", "/drafts", {}) + assert "failed (403)" in str(excinfo.value) + assert "forbidden" in str(excinfo.value) + + +def test_latest_reply_headers_pulls_from_api_get(): + fake_thread = { + "messages": [ + { + "payload": { + "headers": [ + {"name": "Message-ID", "value": "<[email protected]>"}, + ] + } + } + ] + } + with patch("oauth_draft.create_draft.api_get", return_value=fake_thread) as m: + in_reply, refs = latest_reply_headers("token", "thread-id-9") + assert in_reply == "<[email protected]>" + assert refs == "<[email protected]>" + m.assert_called_once_with("token", "/threads/thread-id-9?format=full") + + +def test_create_draft_payload_includes_threadid_when_set(): + raw = b"raw-bytes" + with patch("oauth_draft.create_draft.api_post", return_value={"id": "d-1"}) as m: + result = create_draft("token", "thread-1", raw) + assert result == {"id": "d-1"} + _, _, payload = m.call_args.args + expected_b64 = base64.urlsafe_b64encode(raw).decode().rstrip("=") + assert payload == {"message": {"raw": expected_b64, "threadId": "thread-1"}} + + +def test_create_draft_payload_omits_threadid_when_none(): + with patch("oauth_draft.create_draft.api_post", return_value={"id": "d-2"}) as m: + create_draft("token", None, b"x") + _, _, payload = m.call_args.args + assert "threadId" not in payload["message"] + + +# --- main ------------------------------------------------------------------ + + +def _make_creds_file(tmp_path): + p = tmp_path / "creds.json" + p.write_text( + json.dumps( + { + "client_id": "cid", + "client_secret": "secret", + "refresh_token": "refresh", + "from_address": "[email protected]", + } + ) + ) + return p + + +def test_main_create_draft_end_to_end(tmp_path, capsys): + creds = _make_creds_file(tmp_path) + body_file = tmp_path / "body.txt" + body_file.write_text("Reply body") + api_post_mock = patch( + "oauth_draft.create_draft.api_post", + return_value={ + "id": "draft-id-99", + "message": {"id": "msg-id-99", "threadId": "tid"}, + }, + ) + with ( + patch("oauth_draft.create_draft.refresh_access_token", return_value="tok"), + patch( + "oauth_draft.create_draft.latest_reply_headers", + return_value=("<a@x>", "<a@x>"), + ), + api_post_mock as post, + ): + rc = main( + [ + "--credentials", + str(creds), + "--thread-id", + "tid", + "--to", + "[email protected]", + "--subject", + "Re: hello", + "--body-file", + str(body_file), + ] + ) + assert rc == 0 + out = capsys.readouterr().out + assert "Draft ID: draft-id-99" in out + assert "Message ID: msg-id-99" in out + # Verify the MIME body posted to /drafts has reply headers + body. + _, path, payload = post.call_args.args + assert path == "/drafts" + raw_b64 = payload["message"]["raw"] + raw_bytes = base64.urlsafe_b64decode(raw_b64 + "=" * (-len(raw_b64) % 4)) + decoded = raw_bytes.decode() + assert "From: [email protected]" in decoded + assert "To: [email protected]" in decoded + assert "Subject: Re: hello" in decoded + assert "In-Reply-To: <a@x>" in decoded + assert "Reply body" in decoded + + +def test_main_no_reply_headers_skips_thread_lookup(tmp_path): + creds = _make_creds_file(tmp_path) + body = tmp_path / "b.txt" + body.write_text("x") + with ( + patch("oauth_draft.create_draft.refresh_access_token", return_value="t"), + patch("oauth_draft.create_draft.latest_reply_headers") as latest, + patch( + "oauth_draft.create_draft.api_post", + return_value={"id": "d", "message": {"id": "m"}}, + ), + ): + rc = main( + [ + "--credentials", + str(creds), + "--thread-id", + "tid", + "--no-reply-headers", + "--to", + "[email protected]", + "--subject", + "S", + "--body-file", + str(body), + ] + ) + assert rc == 0 + latest.assert_not_called() diff --git a/tools/gmail/oauth-draft/tests/test_credentials.py b/tools/gmail/oauth-draft/tests/test_credentials.py index 15f794b..d067eb8 100644 --- a/tools/gmail/oauth-draft/tests/test_credentials.py +++ b/tools/gmail/oauth-draft/tests/test_credentials.py @@ -16,11 +16,18 @@ # under the License. from __future__ import annotations +import io import json +import urllib.error +from unittest.mock import patch import pytest -from oauth_draft.credentials import Credentials, locate_credentials +from oauth_draft.credentials import ( + Credentials, + locate_credentials, + refresh_access_token, +) def write_creds(path, **overrides): @@ -92,3 +99,69 @@ def test_locate_raises_when_nothing_exists(tmp_path, monkeypatch): with pytest.raises(SystemExit) as excinfo: locate_credentials(None) assert "No Gmail OAuth credentials found" in str(excinfo.value) + + +# --- refresh_access_token -------------------------------------------------- + + +CREDS = Credentials( + client_id="cid", + client_secret="secret", + refresh_token="refresh", + from_address="[email protected]", +) + + +class _FakeResponse: + """Minimal context-manager stand-in for urllib.request.urlopen().""" + + def __init__(self, payload: bytes): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return self._payload + + +def test_refresh_access_token_returns_access_token(): + with patch("oauth_draft.credentials.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b'{"access_token": "abc-123"}') + token = refresh_access_token(CREDS) + assert token == "abc-123" + # The POST body must include all four OAuth refresh-flow params. + request = mock_open.call_args.args[0] + assert request.method == "POST" + assert request.full_url == "https://oauth2.googleapis.com/token"; + body = request.data.decode() + assert "client_id=cid" in body + assert "client_secret=secret" in body + assert "refresh_token=refresh" in body + assert "grant_type=refresh_token" in body + + +def test_refresh_access_token_raises_on_http_error(): + err = urllib.error.HTTPError( + url="https://oauth2.googleapis.com/token";, + code=400, + msg="Bad Request", + hdrs=None, # type: ignore[arg-type] + fp=io.BytesIO(b'{"error": "invalid_grant"}'), + ) + with patch("oauth_draft.credentials.urllib.request.urlopen", side_effect=err): + with pytest.raises(SystemExit) as excinfo: + refresh_access_token(CREDS) + assert "OAuth token refresh failed (400)" in str(excinfo.value) + assert "invalid_grant" in str(excinfo.value) + + +def test_refresh_access_token_raises_when_no_token_in_payload(): + with patch("oauth_draft.credentials.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b'{"expires_in": 3600}') + with pytest.raises(SystemExit) as excinfo: + refresh_access_token(CREDS) + assert "no access_token" in str(excinfo.value) diff --git a/tools/gmail/oauth-draft/tests/test_mark_threads_read.py b/tools/gmail/oauth-draft/tests/test_mark_threads_read.py index b5ea46a..08326d4 100644 --- a/tools/gmail/oauth-draft/tests/test_mark_threads_read.py +++ b/tools/gmail/oauth-draft/tests/test_mark_threads_read.py @@ -16,7 +16,19 @@ # under the License. from __future__ import annotations -from oauth_draft.mark_threads_read import parse_args +import io +import json +import urllib.error +from unittest.mock import patch + +import pytest + +from oauth_draft.mark_threads_read import ( + list_thread_ids, + main, + modify_thread, + parse_args, +) def test_default_action_is_remove_unread(): @@ -60,3 +72,228 @@ def test_execute_and_max_flags(): args = parse_args(["--query", "x", "--execute", "--max", "5"]) assert args.execute is True assert args.max == 5 + + +# --- network-mocked helpers ------------------------------------------------ + + +class _FakeResponse: + def __init__(self, payload: bytes): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return self._payload + + +def test_list_thread_ids_paginates_until_no_token(): + pages = [ + _FakeResponse( + json.dumps( + { + "threads": [{"id": "t1"}, {"id": "t2"}], + "nextPageToken": "page2", + } + ).encode() + ), + _FakeResponse(json.dumps({"threads": [{"id": "t3"}]}).encode()), + ] + with patch( + "oauth_draft.mark_threads_read.urllib.request.urlopen", + side_effect=pages, + ) as mock_open: + ids = list_thread_ids("token", "in:inbox") + assert ids == ["t1", "t2", "t3"] + # Second call must include the pageToken from the first response. + _, second_call = mock_open.call_args_list + second_url = second_call.args[0].full_url + assert "pageToken=page2" in second_url + + +def test_list_thread_ids_raises_on_http_error(): + err = urllib.error.HTTPError( + url="https://x";, + code=429, + msg="Too Many", + hdrs=None, # type: ignore[arg-type] + fp=io.BytesIO(b'{"error": "rate"}'), + ) + with patch( + "oauth_draft.mark_threads_read.urllib.request.urlopen", + side_effect=err, + ): + with pytest.raises(SystemExit) as excinfo: + list_thread_ids("token", "q") + assert "threads.list failed (429)" in str(excinfo.value) + + +def test_modify_thread_sends_label_payload(): + with patch("oauth_draft.mark_threads_read.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b"{}") + modify_thread("token", "tid-1", ["STARRED"], ["UNREAD"]) + request = mock_open.call_args.args[0] + assert request.method == "POST" + assert "/threads/tid-1/modify" in request.full_url + payload = json.loads(request.data) + assert payload == {"addLabelIds": ["STARRED"], "removeLabelIds": ["UNREAD"]} + + +def test_modify_thread_omits_empty_label_lists(): + with patch("oauth_draft.mark_threads_read.urllib.request.urlopen") as mock_open: + mock_open.return_value = _FakeResponse(b"{}") + modify_thread("token", "tid-2", [], ["UNREAD"]) + payload = json.loads(mock_open.call_args.args[0].data) + assert payload == {"removeLabelIds": ["UNREAD"]} + + +def test_modify_thread_raises_on_http_error(): + err = urllib.error.HTTPError( + url="https://x";, + code=500, + msg="Boom", + hdrs=None, # type: ignore[arg-type] + fp=io.BytesIO(b'{"error": "server"}'), + ) + with patch( + "oauth_draft.mark_threads_read.urllib.request.urlopen", + side_effect=err, + ): + with pytest.raises(SystemExit) as excinfo: + modify_thread("token", "tid", [], ["UNREAD"]) + assert "threads.modify failed (500) for tid" in str(excinfo.value) + + +# --- main ------------------------------------------------------------------ + + +def _make_creds_file(tmp_path): + p = tmp_path / "creds.json" + p.write_text( + json.dumps( + { + "client_id": "cid", + "client_secret": "secret", + "refresh_token": "refresh", + } + ) + ) + return p + + +def test_main_dry_run_lists_ids_and_skips_modify(tmp_path, capsys): + creds = _make_creds_file(tmp_path) + with ( + patch( + "oauth_draft.mark_threads_read.refresh_access_token", + return_value="tok", + ), + patch( + "oauth_draft.mark_threads_read.list_thread_ids", + return_value=["t1", "t2"], + ), + patch("oauth_draft.mark_threads_read.modify_thread") as modify, + ): + rc = main( + [ + "--credentials", + str(creds), + "--query", + "in:inbox is:unread", + ] + ) + assert rc == 0 + modify.assert_not_called() + out = capsys.readouterr().out + assert "t1" in out and "t2" in out + + +def test_main_execute_calls_modify_per_thread(tmp_path): + creds = _make_creds_file(tmp_path) + with ( + patch( + "oauth_draft.mark_threads_read.refresh_access_token", + return_value="tok", + ), + patch( + "oauth_draft.mark_threads_read.list_thread_ids", + return_value=["t1", "t2", "t3"], + ), + patch("oauth_draft.mark_threads_read.modify_thread") as modify, + ): + rc = main( + [ + "--credentials", + str(creds), + "--query", + "q", + "--execute", + ] + ) + assert rc == 0 + assert modify.call_count == 3 + # default action: remove UNREAD + for call in modify.call_args_list: + _, _, add, remove = call.args + assert add == [] + assert remove == ["UNREAD"] + + +def test_main_max_truncates(tmp_path): + creds = _make_creds_file(tmp_path) + with ( + patch( + "oauth_draft.mark_threads_read.refresh_access_token", + return_value="tok", + ), + patch( + "oauth_draft.mark_threads_read.list_thread_ids", + return_value=["t1", "t2", "t3", "t4", "t5"], + ), + patch("oauth_draft.mark_threads_read.modify_thread") as modify, + ): + rc = main( + [ + "--credentials", + str(creds), + "--query", + "q", + "--execute", + "--max", + "2", + ] + ) + assert rc == 0 + assert modify.call_count == 2 + + +def test_main_returns_1_when_modify_fails(tmp_path): + creds = _make_creds_file(tmp_path) + with ( + patch( + "oauth_draft.mark_threads_read.refresh_access_token", + return_value="tok", + ), + patch( + "oauth_draft.mark_threads_read.list_thread_ids", + return_value=["t1"], + ), + patch( + "oauth_draft.mark_threads_read.modify_thread", + side_effect=SystemExit("boom"), + ), + ): + rc = main( + [ + "--credentials", + str(creds), + "--query", + "q", + "--execute", + ] + ) + assert rc == 1 diff --git a/tools/gmail/oauth-draft/tests/test_setup_creds.py b/tools/gmail/oauth-draft/tests/test_setup_creds.py new file mode 100644 index 0000000..3385c2c --- /dev/null +++ b/tools/gmail/oauth-draft/tests/test_setup_creds.py @@ -0,0 +1,219 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import json +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from oauth_draft.setup_creds import detect_from_address, main, parse_args + +# --- detect_from_address --------------------------------------------------- + + +def test_detect_from_address_uses_env_var(monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + assert detect_from_address() == "[email protected]" + + +def test_detect_from_address_falls_back_to_git_config(monkeypatch): + monkeypatch.delenv("GMAIL_FROM", raising=False) + with patch( + "oauth_draft.setup_creds.subprocess.check_output", + return_value="[email protected]\n", + ): + assert detect_from_address() == "[email protected]" + + +def test_detect_from_address_returns_none_when_git_missing(monkeypatch): + monkeypatch.delenv("GMAIL_FROM", raising=False) + with patch( + "oauth_draft.setup_creds.subprocess.check_output", + side_effect=FileNotFoundError, + ): + assert detect_from_address() is None + + +def test_detect_from_address_returns_none_when_git_errors(monkeypatch): + monkeypatch.delenv("GMAIL_FROM", raising=False) + with patch( + "oauth_draft.setup_creds.subprocess.check_output", + side_effect=subprocess.CalledProcessError(1, "git"), + ): + assert detect_from_address() is None + + +def test_detect_from_address_returns_none_when_git_returns_empty(monkeypatch): + monkeypatch.delenv("GMAIL_FROM", raising=False) + with patch( + "oauth_draft.setup_creds.subprocess.check_output", + return_value="\n", + ): + assert detect_from_address() is None + + +# --- parse_args ------------------------------------------------------------ + + +def test_parse_args_minimal(monkeypatch): + # Avoid relying on the host's environment / git config: stub + # detect_from_address out of the parser default. + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + args = parse_args(["client.json"]) + assert args.client_secrets == "client.json" + assert args.from_address == "[email protected]" + assert args.out.endswith("/.config/apache-steward/gmail-oauth.json") + assert args.rm_client_secrets is False + + +def test_parse_args_overrides(monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + args = parse_args( + [ + "client.json", + "--from-address", + "[email protected]", + "--out", + "/custom/path.json", + "--rm-client-secrets", + ] + ) + assert args.from_address == "[email protected]" + assert args.out == "/custom/path.json" + assert args.rm_client_secrets is True + + +# --- main ------------------------------------------------------------------ + + +def _client_secrets(tmp_path): + p = tmp_path / "client_secrets.json" + p.write_text( + json.dumps( + { + "installed": { + "client_id": "cid", + "client_secret": "secret", + } + } + ) + ) + return p + + +def _flow_creds(refresh_token="rt-1"): + creds = MagicMock() + creds.refresh_token = refresh_token + creds.scopes = ["https://mail.google.com/";] + return creds + + +def test_main_errors_when_no_from_address(tmp_path, monkeypatch): + monkeypatch.delenv("GMAIL_FROM", raising=False) + secrets = _client_secrets(tmp_path) + out = tmp_path / "out.json" + with patch( + "oauth_draft.setup_creds.subprocess.check_output", + side_effect=FileNotFoundError, + ): + with pytest.raises(SystemExit) as excinfo: + main([str(secrets), "--out", str(out)]) + assert "Could not determine --from-address" in str(excinfo.value) + + +def test_main_errors_when_client_secrets_missing(tmp_path, monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + out = tmp_path / "out.json" + with pytest.raises(SystemExit) as excinfo: + main([str(tmp_path / "does-not-exist.json"), "--out", str(out)]) + assert "client_secrets not found" in str(excinfo.value) + + +def test_main_errors_when_flow_returns_no_refresh_token(tmp_path, monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + secrets = _client_secrets(tmp_path) + out = tmp_path / "out.json" + flow = MagicMock() + flow.run_local_server.return_value = _flow_creds(refresh_token=None) + with patch( + "oauth_draft.setup_creds.InstalledAppFlow.from_client_secrets_file", + return_value=flow, + ): + with pytest.raises(SystemExit) as excinfo: + main([str(secrets), "--out", str(out)]) + assert "no refresh_token" in str(excinfo.value) + + +def test_main_writes_credentials_file(tmp_path, monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + secrets = _client_secrets(tmp_path) + out_dir = tmp_path / "creds-dir" + out = out_dir / "gmail-oauth.json" + flow = MagicMock() + flow.run_local_server.return_value = _flow_creds() + with patch( + "oauth_draft.setup_creds.InstalledAppFlow.from_client_secrets_file", + return_value=flow, + ): + rc = main([str(secrets), "--out", str(out)]) + assert rc == 0 + written = json.loads(out.read_text()) + assert written == { + "client_id": "cid", + "client_secret": "secret", + "refresh_token": "rt-1", + "from_address": "[email protected]", + } + # Mode should be 600 (owner-rw only) per the atomic-write path. + assert (out.stat().st_mode & 0o777) == 0o600 + # Parent dir should be 700. + assert (out_dir.stat().st_mode & 0o777) == 0o700 + # Original client_secrets is left in place by default. + assert secrets.exists() + + +def test_main_with_rm_client_secrets_deletes_input(tmp_path, monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + secrets = _client_secrets(tmp_path) + out = tmp_path / "creds-dir" / "creds.json" + flow = MagicMock() + flow.run_local_server.return_value = _flow_creds() + with patch( + "oauth_draft.setup_creds.InstalledAppFlow.from_client_secrets_file", + return_value=flow, + ): + main([str(secrets), "--out", str(out), "--rm-client-secrets"]) + assert not secrets.exists() + + +def test_main_handles_web_shaped_client_secrets(tmp_path, monkeypatch): + monkeypatch.setenv("GMAIL_FROM", "[email protected]") + secrets = tmp_path / "client.json" + secrets.write_text(json.dumps({"web": {"client_id": "wcid", "client_secret": "wsecret"}})) + out = tmp_path / "creds-dir" / "creds.json" + flow = MagicMock() + flow.run_local_server.return_value = _flow_creds() + with patch( + "oauth_draft.setup_creds.InstalledAppFlow.from_client_secrets_file", + return_value=flow, + ): + main([str(secrets), "--out", str(out)]) + written = json.loads(out.read_text()) + assert written["client_id"] == "wcid" + assert written["client_secret"] == "wsecret" diff --git a/tools/vulnogram/generate-cve-json/.gitignore b/tools/vulnogram/generate-cve-json/.gitignore index ab7ef44..8cae406 100644 --- a/tools/vulnogram/generate-cve-json/.gitignore +++ b/tools/vulnogram/generate-cve-json/.gitignore @@ -4,3 +4,4 @@ __pycache__/ .pytest_cache/ .ruff_cache/ .mypy_cache/ +.coverage
