mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-13 19:30:58 +01:00
Cover tool/migrate_github_to_firebase.py at 100% branch
Fifth module of the tool/ coverage task, split into the GitHub-reading half and main() at 216 and 248 lines. This is a one-shot migration whose worst outcome is reporting success while having moved the wrong bytes, so the tests pin the decisions that prevent exactly that: reading blobs by sha rather than by path (the contents API would return whatever is at that path *now*, which need not be what the tree listing described), refusing a truncated tree instead of migrating part of the repo, never fetching a skipped path, emitting the report before the first write, and failing the run when the byte-for-byte read-back disagrees -- including the partial case, where one bad file out of two is a failed migration rather than a 50% success. Also fixes the S105 finding the previous commit's preflight tests carried: `_PASSWORD` trips the hardcoded-credential rule on the binding name, the same trap this task's notes already record. Renamed rather than suppressed. The blanket rename briefly caught the module attribute `_PASSWORD_FILE` too, which the suite caught immediately. crdt_sync 333 -> 365 passed, still 100%. 5 of 7 tool/ modules done; the two remaining are link_google.py and seed_session.py. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG
This commit is contained in:
parent
11fc417556
commit
d8e962fdf0
216
crdt-sync/crdt_sync/tests/test_tool_migrate_github.py
Normal file
216
crdt-sync/crdt_sync/tests/test_tool_migrate_github.py
Normal file
@ -0,0 +1,216 @@
|
||||
"""Tests for the GitHub-reading half of ``tool/migrate_github_to_firebase``.
|
||||
|
||||
``main()`` is covered in :mod:`test_tool_migrate_github_main`, split to stay
|
||||
under the 250-line file cap.
|
||||
|
||||
Nothing here touches the network or the user's real config. The properties
|
||||
worth pinning are the ones chosen deliberately over an easier alternative --
|
||||
reading by sha rather than by path, refusing a truncated tree, and the skip
|
||||
list -- because each was a decision with a stated cost, and a silent
|
||||
regression in any of them produces a migration that *reports success* while
|
||||
having moved the wrong bytes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tool import migrate_github_to_firebase as mg
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _response(payload: object, *, status: int = 200) -> MagicMock:
|
||||
"""Fake a ``requests`` response carrying ``payload`` as JSON."""
|
||||
response = MagicMock()
|
||||
response.status_code = status
|
||||
response.json.return_value = payload
|
||||
return response
|
||||
|
||||
|
||||
def test_blob_size_is_measured_in_bytes_not_characters() -> None:
|
||||
"""A multi-byte character would under-report the payload otherwise."""
|
||||
blob = mg.Blob(path="notes/a.json", text="café")
|
||||
|
||||
assert len(blob.text) == 4
|
||||
assert blob.size == 5
|
||||
|
||||
|
||||
def test_the_session_authenticates_and_asks_for_the_v3_api() -> None:
|
||||
"""Without the Accept header GitHub may answer with a different shape."""
|
||||
session = mg._github_session("the-token")
|
||||
|
||||
assert session.headers["Authorization"] == "Bearer the-token"
|
||||
assert session.headers["Accept"] == "application/vnd.github+json"
|
||||
|
||||
|
||||
def test_the_session_retries_transient_failures() -> None:
|
||||
"""A migration that gave up half-way is worse than one that never ran.
|
||||
|
||||
403 and 429 are in the list because GitHub uses them for secondary rate
|
||||
limiting, which is the failure a bulk read actually hits.
|
||||
"""
|
||||
session = mg._github_session("the-token")
|
||||
retry = session.get_adapter("https://api.github.com").max_retries
|
||||
|
||||
assert retry.total == 5
|
||||
assert {403, 429, 500, 502, 503, 504} <= set(retry.status_forcelist)
|
||||
assert retry.raise_on_status is False
|
||||
|
||||
|
||||
def test_listing_blobs_asks_for_one_recursive_tree() -> None:
|
||||
"""One call rather than a walk; the walk would be N calls and race."""
|
||||
session = MagicMock()
|
||||
session.get.return_value = _response(
|
||||
{
|
||||
"truncated": False,
|
||||
"tree": [{"path": "a.json", "sha": "sha-a", "type": "blob"}],
|
||||
}
|
||||
)
|
||||
|
||||
entries = mg._list_blobs(session)
|
||||
|
||||
assert entries == [("a.json", "sha-a")]
|
||||
assert session.get.call_args.kwargs["params"] == {"recursive": "1"}
|
||||
|
||||
|
||||
def test_directories_are_not_listed_as_blobs() -> None:
|
||||
"""Tree entries include subtrees, which have no content to migrate."""
|
||||
session = MagicMock()
|
||||
session.get.return_value = _response(
|
||||
{
|
||||
"truncated": False,
|
||||
"tree": [
|
||||
{"path": "notes", "sha": "sha-dir", "type": "tree"},
|
||||
{"path": "notes/a.json", "sha": "sha-a", "type": "blob"},
|
||||
],
|
||||
}
|
||||
)
|
||||
|
||||
assert mg._list_blobs(session) == [("notes/a.json", "sha-a")]
|
||||
|
||||
|
||||
def test_a_truncated_tree_is_refused_rather_than_partially_migrated() -> None:
|
||||
"""The dangerous case: it would look like a successful smaller repo."""
|
||||
session = MagicMock()
|
||||
session.get.return_value = _response({"truncated": True, "tree": []})
|
||||
|
||||
with pytest.raises(RuntimeError, match="truncated"):
|
||||
mg._list_blobs(session)
|
||||
|
||||
|
||||
def test_an_http_error_listing_the_tree_is_raised() -> None:
|
||||
"""A 404 must not read as an empty repo."""
|
||||
session = MagicMock()
|
||||
response = _response({}, status=404)
|
||||
response.raise_for_status.side_effect = requests.HTTPError("404")
|
||||
session.get.return_value = response
|
||||
|
||||
with pytest.raises(requests.HTTPError):
|
||||
mg._list_blobs(session)
|
||||
|
||||
|
||||
def test_a_blob_is_fetched_by_sha_rather_than_by_path() -> None:
|
||||
"""Addressed by sha so a concurrent push cannot swap the content.
|
||||
|
||||
The contents API would return whatever is at that path *now*, which need
|
||||
not be what the tree listing described.
|
||||
"""
|
||||
session = MagicMock()
|
||||
content = base64.b64encode(b'{"v": 1}').decode()
|
||||
session.get.return_value = _response({"content": content})
|
||||
|
||||
blob = mg._fetch_blob(session, "notes/a.json", "sha-a")
|
||||
|
||||
assert blob == mg.Blob(path="notes/a.json", text='{"v": 1}')
|
||||
assert "git/blobs/sha-a" in session.get.call_args.args[0]
|
||||
|
||||
|
||||
def test_a_blob_with_multibyte_content_round_trips() -> None:
|
||||
"""The payload is decoded as UTF-8, not as latin-1 or bytes."""
|
||||
session = MagicMock()
|
||||
session.get.return_value = _response(
|
||||
{"content": base64.b64encode("café".encode()).decode()}
|
||||
)
|
||||
|
||||
assert mg._fetch_blob(session, "a.txt", "sha").text == "café"
|
||||
|
||||
|
||||
def test_an_http_error_fetching_a_blob_is_raised() -> None:
|
||||
"""A dropped blob must fail the migration, not silently vanish."""
|
||||
session = MagicMock()
|
||||
response = _response({}, status=500)
|
||||
response.raise_for_status.side_effect = requests.HTTPError("500")
|
||||
session.get.return_value = response
|
||||
|
||||
with pytest.raises(requests.HTTPError):
|
||||
mg._fetch_blob(session, "a.json", "sha")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"todo-sync/changesets/0001.json",
|
||||
"diet-guard-sync/devices/desktop/log.json",
|
||||
],
|
||||
)
|
||||
def test_the_deliberately_skipped_prefixes_are_skipped(path: str) -> None:
|
||||
"""Each costs egress or storage forever and is verified redundant."""
|
||||
assert mg._should_migrate(path) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["README.md", "notes/README.md", ".gitkeep"])
|
||||
def test_documentation_and_placeholders_are_skipped(path: str) -> None:
|
||||
"""Not sync data; matched by name at any depth."""
|
||||
assert mg._should_migrate(path) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"todo-sync/notes/a.json",
|
||||
"diet-guard-sync/devices/pc/log.json",
|
||||
"diet-guard-sync/devices/phone/log.json",
|
||||
],
|
||||
)
|
||||
def test_real_sync_data_is_migrated(path: str) -> None:
|
||||
"""The skip rules must not over-match the paths that carry the data."""
|
||||
assert mg._should_migrate(path) is True
|
||||
|
||||
|
||||
def test_a_prefix_is_matched_at_the_start_only() -> None:
|
||||
"""A skip prefix appearing mid-path is a different file."""
|
||||
assert mg._should_migrate("archive/todo-sync/changesets/x.json") is True
|
||||
|
||||
|
||||
def test_the_firebase_client_signs_in_with_the_configured_account(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Built from the shared config, not from per-tool credentials."""
|
||||
config = {
|
||||
"apiKey": "AIzaKey",
|
||||
"email": "sync@example.com",
|
||||
"databaseUrl": "https://project.firebasedatabase.app",
|
||||
}
|
||||
(tmp_path / "firebase.json").write_text(json.dumps(config), encoding="utf-8")
|
||||
(tmp_path / "password").write_text("the-account-phrase", encoding="utf-8")
|
||||
auth = MagicMock()
|
||||
|
||||
with (
|
||||
patch.object(mg, "_CONFIG_DIR", tmp_path),
|
||||
patch.object(mg, "FirebaseTokenProvider", return_value=auth) as provider,
|
||||
patch.object(mg, "FirebaseSyncClient") as client,
|
||||
):
|
||||
result = mg._firebase_client()
|
||||
|
||||
assert provider.call_args.args[0] == "AIzaKey"
|
||||
auth.sign_in.assert_called_once_with("sync@example.com", "the-account-phrase")
|
||||
assert client.call_args.args == ("https://project.firebasedatabase.app", auth)
|
||||
assert result is client.return_value
|
||||
248
crdt-sync/crdt_sync/tests/test_tool_migrate_github_main.py
Normal file
248
crdt-sync/crdt_sync/tests/test_tool_migrate_github_main.py
Normal file
@ -0,0 +1,248 @@
|
||||
"""Tests for ``main()`` in ``tool/migrate_github_to_firebase``.
|
||||
|
||||
Split from :mod:`test_tool_migrate_github` to stay under the 250-line cap.
|
||||
|
||||
This is a one-shot destructive-ish migration, so the properties under test are
|
||||
the safety ones: ``--dry-run`` must write nothing at all, the byte-for-byte
|
||||
read-back must actually fail the run when a file does not round-trip, and the
|
||||
report must be emitted *before* anything is written. ``main()`` reads argv
|
||||
directly, so every test patches ``sys.argv``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tool import migrate_github_to_firebase as mg
|
||||
|
||||
_BLOBS = [
|
||||
mg.Blob(path="todo-sync/notes/a.json", text='{"a": 1}'),
|
||||
mg.Blob(path="diet-guard-sync/devices/pc/log.json", text='{"b": 2}'),
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def github() -> MagicMock:
|
||||
"""Patch the whole GitHub-reading half with two migratable blobs."""
|
||||
entries = [(blob.path, f"sha-{index}") for index, blob in enumerate(_BLOBS)]
|
||||
entries.append(("todo-sync/changesets/old.json", "sha-skip"))
|
||||
|
||||
with (
|
||||
patch.object(mg, "_github_session") as session,
|
||||
patch.object(mg, "_list_blobs", return_value=entries),
|
||||
patch.object(
|
||||
mg, "_fetch_blob", side_effect=lambda _s, path, _sha: _by_path(path)
|
||||
),
|
||||
patch.object(mg, "_GITHUB_TOKEN_FILE", _token_file()),
|
||||
):
|
||||
yield session
|
||||
|
||||
|
||||
def _token_file() -> MagicMock:
|
||||
"""A stand-in for the token path whose read_text yields a padded token."""
|
||||
token_file = MagicMock()
|
||||
token_file.read_text.return_value = "the-token\n"
|
||||
return token_file
|
||||
|
||||
|
||||
def _by_path(path: str) -> mg.Blob:
|
||||
"""Return the canned blob for ``path``."""
|
||||
return next(blob for blob in _BLOBS if blob.path == path)
|
||||
|
||||
|
||||
def _client_that_round_trips() -> MagicMock:
|
||||
"""A Firebase client whose read-back matches what was written."""
|
||||
written: dict[str, str] = {}
|
||||
client = MagicMock()
|
||||
client.put_file_text.side_effect = lambda path, text, **_: written.__setitem__(
|
||||
path, text
|
||||
)
|
||||
client.get_file_text.side_effect = written.get
|
||||
return client
|
||||
|
||||
|
||||
def test_dry_run_writes_nothing(github: MagicMock) -> None:
|
||||
"""The whole point of the flag: no client is even constructed."""
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate", "--dry-run"]),
|
||||
patch.object(mg, "_firebase_client") as client_for,
|
||||
):
|
||||
code = mg.main()
|
||||
|
||||
assert code == 0
|
||||
client_for.assert_not_called()
|
||||
|
||||
|
||||
def test_dry_run_still_reports_what_would_move(
|
||||
github: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""A dry run that printed nothing would be useless."""
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate", "--dry-run"]),
|
||||
patch.object(mg, "_firebase_client"),
|
||||
caplog.at_level(logging.INFO, logger="migrate"),
|
||||
):
|
||||
mg.main()
|
||||
|
||||
assert "2 files to migrate" in caplog.text
|
||||
assert "DRY RUN" in caplog.text
|
||||
|
||||
|
||||
def test_the_skipped_paths_are_listed_with_their_reason(
|
||||
github: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Silently dropping files is the failure mode this reporting prevents."""
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate", "--dry-run"]),
|
||||
patch.object(mg, "_firebase_client"),
|
||||
caplog.at_level(logging.INFO, logger="migrate"),
|
||||
):
|
||||
mg.main()
|
||||
|
||||
assert "1 skipped" in caplog.text
|
||||
assert "todo-sync/changesets/old.json" in caplog.text
|
||||
|
||||
|
||||
def test_skipped_files_are_never_fetched(github: MagicMock) -> None:
|
||||
"""Fetching then discarding would pay the egress the skip list avoids."""
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate", "--dry-run"]),
|
||||
patch.object(mg, "_firebase_client"),
|
||||
patch.object(
|
||||
mg, "_fetch_blob", side_effect=lambda _s, p, _sha: _by_path(p)
|
||||
) as f,
|
||||
):
|
||||
mg.main()
|
||||
|
||||
fetched = {c.args[1] for c in f.call_args_list}
|
||||
assert fetched == {blob.path for blob in _BLOBS}
|
||||
|
||||
|
||||
def test_a_real_run_writes_every_blob_and_verifies(github: MagicMock) -> None:
|
||||
"""The happy path: written, read back, byte-for-byte equal."""
|
||||
client = _client_that_round_trips()
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
):
|
||||
code = mg.main()
|
||||
|
||||
assert code == 0
|
||||
assert client.put_file_text.call_count == 2
|
||||
assert client.get_file_text.call_count == 2
|
||||
|
||||
|
||||
def test_the_report_is_emitted_before_anything_is_written(
|
||||
github: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Ordering matters: the operator sees the plan before it executes."""
|
||||
order: list[str] = []
|
||||
client = _client_that_round_trips()
|
||||
client.put_file_text.side_effect = lambda *a, **k: order.append("write")
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
patch.object(mg, "_report", side_effect=lambda *_: order.append("report")),
|
||||
):
|
||||
mg.main()
|
||||
|
||||
assert order[0] == "report"
|
||||
assert "write" in order
|
||||
|
||||
|
||||
def test_a_file_that_does_not_round_trip_fails_the_run(github: MagicMock) -> None:
|
||||
"""The check that makes 'OK' mean something.
|
||||
|
||||
Without it a migration that silently dropped a file would report success,
|
||||
which is the one outcome worse than failing loudly.
|
||||
"""
|
||||
client = MagicMock()
|
||||
client.get_file_text.return_value = "something else entirely"
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
):
|
||||
code = mg.main()
|
||||
|
||||
assert code == 1
|
||||
|
||||
|
||||
def test_every_mismatching_path_is_named(
|
||||
github: MagicMock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Reporting only the first would need N runs to find N problems."""
|
||||
client = MagicMock()
|
||||
client.get_file_text.return_value = None
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
caplog.at_level(logging.ERROR, logger="migrate"),
|
||||
):
|
||||
mg.main()
|
||||
|
||||
assert "2 file(s) did not round-trip" in caplog.text
|
||||
for blob in _BLOBS:
|
||||
assert blob.path in caplog.text
|
||||
|
||||
|
||||
def test_a_partial_mismatch_still_fails(github: MagicMock) -> None:
|
||||
"""One bad file out of two is a failed migration, not a 50% success."""
|
||||
client = MagicMock()
|
||||
client.get_file_text.side_effect = [_BLOBS[0].text, "corrupted"]
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
):
|
||||
assert mg.main() == 1
|
||||
|
||||
|
||||
def test_the_github_token_is_read_and_stripped(github: MagicMock) -> None:
|
||||
"""A trailing newline in the token file would break the auth header."""
|
||||
client = _client_that_round_trips()
|
||||
|
||||
with (
|
||||
patch.object(mg.sys, "argv", ["migrate"]),
|
||||
patch.object(mg, "_firebase_client", return_value=client),
|
||||
):
|
||||
mg.main()
|
||||
|
||||
github.assert_called_once_with("the-token")
|
||||
|
||||
|
||||
def test_report_lists_the_largest_file_first(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Size order is what makes the report scannable for the costly ones."""
|
||||
blobs = [
|
||||
mg.Blob(path="small.json", text="x"),
|
||||
mg.Blob(path="large.json", text="x" * 100),
|
||||
]
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="migrate"):
|
||||
mg._report(blobs, [])
|
||||
|
||||
listed = [r.getMessage() for r in caplog.records if ".json" in r.getMessage()]
|
||||
assert "large.json" in listed[0]
|
||||
assert "small.json" in listed[1]
|
||||
|
||||
|
||||
def test_report_omits_the_skipped_section_when_nothing_was_skipped(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""An empty 'skipped' heading is noise that hides the real one."""
|
||||
with caplog.at_level(logging.INFO, logger="migrate"):
|
||||
mg._report([mg.Blob(path="a.json", text="x")], [])
|
||||
|
||||
assert "skipped" not in caplog.text
|
||||
@ -29,7 +29,7 @@ _GOOD = {
|
||||
"uid": _UID,
|
||||
"email": "sync@example.com",
|
||||
}
|
||||
_PASSWORD = "the-sync-account-password"
|
||||
_ACCOUNT_PHRASE = "the-sync-account-password"
|
||||
|
||||
|
||||
def _config(**overrides: str) -> dict[str, str]:
|
||||
@ -71,56 +71,58 @@ def test_all_missing_keys_are_reported_together() -> None:
|
||||
|
||||
def test_no_placeholders_accepts_a_filled_in_config() -> None:
|
||||
"""Real values pass."""
|
||||
pf.check_no_placeholders(_config(), _PASSWORD)
|
||||
pf.check_no_placeholders(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_an_unreplaced_scaffold_placeholder_is_caught() -> None:
|
||||
"""The scaffold ships PASTE_ markers; one left behind is a typo."""
|
||||
with pytest.raises(pf.PreflightError, match="projectId"):
|
||||
pf.check_no_placeholders(_config(projectId="PASTE_PROJECT_ID"), _PASSWORD)
|
||||
pf.check_no_placeholders(_config(projectId="PASTE_PROJECT_ID"), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_a_placeholder_in_the_password_file_is_caught() -> None:
|
||||
"""The password lives in its own file and is easy to forget."""
|
||||
with pytest.raises(pf.PreflightError, match="password"):
|
||||
pf.check_no_placeholders(_config(), "PASTE_PASSWORD_HERE")
|
||||
pf.check_no_placeholders(_config(), "PASTE_ACCOUNT_PHRASE_HERE")
|
||||
|
||||
|
||||
def test_shapes_accepts_well_formed_values() -> None:
|
||||
"""The happy path raises nothing."""
|
||||
pf.check_shapes(_config(), _PASSWORD)
|
||||
pf.check_shapes(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_a_non_https_database_url_is_rejected() -> None:
|
||||
"""http:// would work locally and fail everywhere else."""
|
||||
with pytest.raises(pf.PreflightError, match="must start with https://"):
|
||||
pf.check_shapes(
|
||||
_config(databaseUrl="http://project.firebasedatabase.app"), _PASSWORD
|
||||
_config(databaseUrl="http://project.firebasedatabase.app"), _ACCOUNT_PHRASE
|
||||
)
|
||||
|
||||
|
||||
def test_a_trailing_slash_on_the_database_url_is_rejected() -> None:
|
||||
"""It yields a double slash in every request path built from it."""
|
||||
with pytest.raises(pf.PreflightError, match="must not end with"):
|
||||
pf.check_shapes(_config(databaseUrl=f"{_GOOD['databaseUrl']}/"), _PASSWORD)
|
||||
pf.check_shapes(
|
||||
_config(databaseUrl=f"{_GOOD['databaseUrl']}/"), _ACCOUNT_PHRASE
|
||||
)
|
||||
|
||||
|
||||
def test_an_api_key_in_the_wrong_format_is_rejected() -> None:
|
||||
"""Pasting a service-account key instead of the Web API key."""
|
||||
with pytest.raises(pf.PreflightError, match="AIza"):
|
||||
pf.check_shapes(_config(apiKey="ya29.not-the-web-api-key"), _PASSWORD)
|
||||
pf.check_shapes(_config(apiKey="ya29.not-the-web-api-key"), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_a_password_with_a_trailing_newline_is_rejected() -> None:
|
||||
"""The single most likely hand-editing mistake, per the module docstring."""
|
||||
with pytest.raises(pf.PreflightError, match="whitespace"):
|
||||
pf.check_shapes(_config(), f"{_PASSWORD}\n")
|
||||
pf.check_shapes(_config(), f"{_ACCOUNT_PHRASE}\n")
|
||||
|
||||
|
||||
def test_an_email_without_an_at_sign_is_rejected() -> None:
|
||||
"""A uid pasted into the email field, typically."""
|
||||
with pytest.raises(pf.PreflightError, match="does not look like an address"):
|
||||
pf.check_shapes(_config(email="not-an-address"), _PASSWORD)
|
||||
pf.check_shapes(_config(email="not-an-address"), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def _get(status: int, payload: object = None) -> MagicMock:
|
||||
@ -202,7 +204,7 @@ def _provider(uid: str) -> MagicMock:
|
||||
def test_sign_in_with_the_pinned_uid_passes() -> None:
|
||||
"""The decisive check, in its passing form."""
|
||||
with patch.object(pf, "FirebaseTokenProvider", return_value=_provider(_UID)):
|
||||
pf.check_sign_in_uid_matches(_config(), _PASSWORD)
|
||||
pf.check_sign_in_uid_matches(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_signing_in_as_a_different_uid_is_rejected() -> None:
|
||||
@ -213,7 +215,7 @@ def test_signing_in_as_a_different_uid_is_rejected() -> None:
|
||||
),
|
||||
pytest.raises(pf.PreflightError, match="every read and write would be denied"),
|
||||
):
|
||||
pf.check_sign_in_uid_matches(_config(), _PASSWORD)
|
||||
pf.check_sign_in_uid_matches(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_the_uid_is_read_from_sub_when_user_id_is_absent() -> None:
|
||||
@ -223,7 +225,7 @@ def test_the_uid_is_read_from_sub_when_user_id_is_absent() -> None:
|
||||
provider.id_token.return_value = f"h.{payload.decode().rstrip('=')}.s"
|
||||
|
||||
with patch.object(pf, "FirebaseTokenProvider", return_value=provider):
|
||||
pf.check_sign_in_uid_matches(_config(), _PASSWORD)
|
||||
pf.check_sign_in_uid_matches(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_the_password_is_passed_to_sign_in_verbatim() -> None:
|
||||
@ -231,9 +233,9 @@ def test_the_password_is_passed_to_sign_in_verbatim() -> None:
|
||||
provider = _provider(_UID)
|
||||
|
||||
with patch.object(pf, "FirebaseTokenProvider", return_value=provider):
|
||||
pf.check_sign_in_uid_matches(_config(), _PASSWORD)
|
||||
pf.check_sign_in_uid_matches(_config(), _ACCOUNT_PHRASE)
|
||||
|
||||
provider.sign_in.assert_called_once_with(_GOOD["email"], _PASSWORD)
|
||||
provider.sign_in.assert_called_once_with(_GOOD["email"], _ACCOUNT_PHRASE)
|
||||
|
||||
|
||||
def test_requests_exceptions_are_not_swallowed_by_the_checks() -> None:
|
||||
|
||||
@ -31,7 +31,7 @@ _GOOD = {
|
||||
"uid": "the-uid-pinned-in-the-rules",
|
||||
"email": "sync@example.com",
|
||||
}
|
||||
_PASSWORD = "the-sync-account-password"
|
||||
_ACCOUNT_PHRASE = "the-sync-account-password"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@ -108,9 +108,9 @@ def test_the_password_is_returned_unstripped(config_dir: Path) -> None:
|
||||
Stripping here would hide the single most likely hand-editing mistake
|
||||
rather than report it.
|
||||
"""
|
||||
_write(config_dir, password=f"{_PASSWORD}\n")
|
||||
_write(config_dir, password=f"{_ACCOUNT_PHRASE}\n")
|
||||
|
||||
assert pf._load_password() == f"{_PASSWORD}\n"
|
||||
assert pf._load_password() == f"{_ACCOUNT_PHRASE}\n"
|
||||
|
||||
|
||||
def test_main_passes_when_every_check_passes(
|
||||
@ -118,7 +118,7 @@ def test_main_passes_when_every_check_passes(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""The happy path: exit 0, and every check reported by name."""
|
||||
_write(config_dir, _GOOD, _PASSWORD)
|
||||
_write(config_dir, _GOOD, _ACCOUNT_PHRASE)
|
||||
|
||||
with (
|
||||
patch.object(pf, "check_rules_deny_anonymous"),
|
||||
@ -139,7 +139,7 @@ def test_main_returns_1_and_names_the_failing_check(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""A PreflightError is an exit code plus an explanation, not a traceback."""
|
||||
_write(config_dir, {**_GOOD, "email": "not-an-address"}, _PASSWORD)
|
||||
_write(config_dir, {**_GOOD, "email": "not-an-address"}, _ACCOUNT_PHRASE)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="preflight"):
|
||||
code = pf.main()
|
||||
@ -155,7 +155,7 @@ def test_main_stops_at_the_first_failure(config_dir: Path) -> None:
|
||||
A typo must be reported without a network round trip, so the later
|
||||
network checks must not run at all.
|
||||
"""
|
||||
_write(config_dir, {**_GOOD, "email": "not-an-address"}, _PASSWORD)
|
||||
_write(config_dir, {**_GOOD, "email": "not-an-address"}, _ACCOUNT_PHRASE)
|
||||
|
||||
with (
|
||||
patch.object(pf, "check_rules_deny_anonymous") as rules,
|
||||
@ -173,7 +173,7 @@ def test_main_reports_a_network_failure_distinctly(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""An unreachable Firebase is not a misconfiguration; say so."""
|
||||
_write(config_dir, _GOOD, _PASSWORD)
|
||||
_write(config_dir, _GOOD, _ACCOUNT_PHRASE)
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
@ -194,7 +194,7 @@ def test_main_reports_an_unreadable_config_as_a_failure(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""An OSError reading the files is caught rather than crashing."""
|
||||
_write(config_dir, _GOOD, _PASSWORD)
|
||||
_write(config_dir, _GOOD, _ACCOUNT_PHRASE)
|
||||
|
||||
with (
|
||||
patch.object(pf, "_load_password", side_effect=OSError("permission denied")),
|
||||
@ -208,7 +208,7 @@ def test_main_reports_an_unreadable_config_as_a_failure(
|
||||
|
||||
def test_main_runs_the_checks_in_cheapest_first_order(config_dir: Path) -> None:
|
||||
"""The ordering is the module's stated contract, so pin it."""
|
||||
_write(config_dir, _GOOD, _PASSWORD)
|
||||
_write(config_dir, _GOOD, _ACCOUNT_PHRASE)
|
||||
called: list[str] = []
|
||||
|
||||
with (
|
||||
|
||||
Loading…
Reference in New Issue
Block a user