"""Unit: the git-source token schema surface (phase 121, task 01). Task 01 lands the STORAGE only: the write-side input shapes can carry the masked credential, and the output shapes structurally cannot. * ``GitSourceIn.token`` / ``GitSourcePatchIn.token`` — optional ``str | None`` (absent/None = no credential / no change), trimmed *before* the max-500 length constraint runs (the ``_trim_url`` precedent), ``None`` passing through untouched; * ``GitSourcePatchIn.token`` — the tri-state documented contract: absent/None = no change, non-empty = replace, empty string = clear; * ``GitSourceOut`` / ``GitSourceRow`` — NO ``token`` field (LOCKED A2): the credential never reaches the UI or any API output, and ``extra="forbid"`` makes the omission structural — constructing an output model with a ``token`` key (kwarg OR dict) raises, so a regression that tries to echo the credential back cannot even build the shape (task 04 finalizes this file with the E2E-level pins). Task 02 lands the CLONE + SANITIZATION mechanics (pure helpers, no DB): * ``sanitize_url`` — strips the userinfo of ``https?://`` URLs (``user:pass@`` and the username-as-token form), leaves ``ssh://`` / ``git@`` / local paths untouched, is idempotent, and is byte-identical for credential-free URLs (anchored regex, never a URL parser re-serialization); * ``clone_url_for`` — NULL/falsy token → the bare stored URL verbatim (public + legacy rows clone exactly as pre-phase), https? row with a token → ``https://x-access-token:@…`` (any existing userinfo replaced by the column credential), non-https row with a token → the URL unchanged + a warning log (no crash); * ``normalize_credential`` — embedded userinfo moves to the token column (password part of ``user:pass``; the whole run for the username-as-token form), an explicit token (even "") wins (LOCKED A6), clean URLs + ssh/``git@``/local paths come back untouched. """ from __future__ import annotations import re import uuid from pathlib import Path import pytest from pydantic import ValidationError from app.models import GitSource from app.rag.git_sources import clone_url_for, normalize_credential, sanitize_url from app.schemas import GitSourceIn, GitSourceOut, GitSourcePatchIn, GitSourceRow URL = "https://github.com/owner/repo.git" # --------------------------------------------------------------------------- # GitSourceIn — write side: accepts + trims the token # --------------------------------------------------------------------------- def test_git_source_in_token_defaults_to_none() -> None: """Absent token = no credential (public repo) — the pre-phase-121 body shape still validates unchanged.""" payload = GitSourceIn(url=URL) assert payload.token is None def test_git_source_in_token_accepted() -> None: payload = GitSourceIn(url=URL, token="ghp_secret123") assert payload.token == "ghp_secret123" def test_git_source_in_token_trimmed_before_length_constraints() -> None: """The ``_trim_url`` precedent: surrounding whitespace is stripped before the max-500 constraint runs.""" payload = GitSourceIn(url=URL, token=" ghp_secret123 ") assert payload.token == "ghp_secret123" def test_git_source_in_token_whitespace_only_becomes_empty() -> None: """Trimmed to ``""`` — valid (create-time has no min_length): the caller sent a blank masked field, i.e. no credential.""" payload = GitSourceIn(url=URL, token=" ") assert payload.token == "" def test_git_source_in_token_at_cap_validates() -> None: token = "x" * 500 payload = GitSourceIn(url=URL, token=token) assert payload.token == token def test_git_source_in_token_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: GitSourceIn(url=URL, token="x" * 501) assert exc.value.errors()[0]["loc"] == ("token",) def test_git_source_in_pre_phase_fields_still_validate() -> None: """The pre-phase-121 body (url + ignore_paths + include_hidden) is unchanged — the new field is purely additive.""" payload = GitSourceIn(url=URL, ignore_paths=["docs/private"], include_hidden=True) assert payload.url == URL assert payload.ignore_paths == ["docs/private"] assert payload.include_hidden is True assert payload.token is None # --------------------------------------------------------------------------- # GitSourcePatchIn — tri-state: absent/None no change, non-empty replace, # empty string clear # --------------------------------------------------------------------------- def test_git_source_patch_in_token_defaults_to_none() -> None: """Absent/None = NO CHANGE — the row's stored credential survives an edit that does not touch the masked field (the edit modal sends it blank → the router omits the key → None here).""" payload = GitSourcePatchIn() assert payload.token is None def test_git_source_patch_in_token_replace_value() -> None: payload = GitSourcePatchIn(token="new-secret") assert payload.token == "new-secret" def test_git_source_patch_in_token_trimmed() -> None: payload = GitSourcePatchIn(token=" new-secret ") assert payload.token == "new-secret" def test_git_source_patch_in_token_empty_string_clears() -> None: """Empty string = CLEAR (the UI offers replace; clear exists for API completeness).""" payload = GitSourcePatchIn(token="") assert payload.token == "" def test_git_source_patch_in_token_whitespace_only_clears() -> None: """Whitespace-only trims to the clear value — a pasted space in the masked field is a clear, not a five-character credential.""" payload = GitSourcePatchIn(token=" ") assert payload.token == "" def test_git_source_patch_in_token_at_cap_validates() -> None: token = "x" * 500 payload = GitSourcePatchIn(token=token) assert payload.token == token def test_git_source_patch_in_token_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: GitSourcePatchIn(token="x" * 501) assert exc.value.errors()[0]["loc"] == ("token",) def test_git_source_patch_in_other_fields_unaffected() -> None: """ignore_paths / include_hidden keep their independent optionality — the new field adds a third state, not a coupling.""" payload = GitSourcePatchIn(ignore_paths=["a"], include_hidden=True) assert payload.ignore_paths == ["a"] assert payload.include_hidden is True assert payload.token is None # --------------------------------------------------------------------------- # GitSourceOut / GitSourceRow — NO token field, ever (LOCKED A2) # --------------------------------------------------------------------------- def _out_kwargs() -> dict: return { "id": uuid.uuid4(), "url": URL, "added_at": None, "ignore_paths": [], "include_hidden": False, } def _row_kwargs() -> dict: return { "id": uuid.uuid4(), "kind": "git", "url": URL, "path": None, "added_at": None, "ignore_paths": [], "include_hidden": False, } def test_output_models_have_no_token_field() -> None: """The omission is a documented contract, not an accident: neither response shape declares a token field at all.""" assert "token" not in GitSourceOut.model_fields assert "token" not in GitSourceRow.model_fields def test_git_source_out_rejects_token_kwarg() -> None: """``extra="forbid"`` — a regression that tries to echo the stored credential back cannot even build the shape. (The kwarg is a deliberate type error — pyright statically knows the shape has no token parameter; that is the contract under test.)""" with pytest.raises(ValidationError) as exc: GitSourceOut(token="ghp_must_never_leak", **_out_kwargs()) # type: ignore[reportCallIssue] assert ("token",) in [tuple(e["loc"]) for e in exc.value.errors()] def test_git_source_out_rejects_token_key() -> None: with pytest.raises(ValidationError) as exc: GitSourceOut.model_validate({**_out_kwargs(), "token": "ghp_must_never_leak"}) assert ("token",) in [tuple(e["loc"]) for e in exc.value.errors()] def test_git_source_row_rejects_token_kwarg() -> None: with pytest.raises(ValidationError) as exc: # Same deliberate type error as the GitSourceOut pin above — the # shape has no token parameter (pyright knows; runtime forbids). GitSourceRow(token="ghp_must_never_leak", **_row_kwargs()) # type: ignore[reportCallIssue] assert ("token",) in [tuple(e["loc"]) for e in exc.value.errors()] def test_git_source_row_rejects_token_key() -> None: with pytest.raises(ValidationError) as exc: GitSourceRow.model_validate({**_row_kwargs(), "token": "ghp_must_never_leak"}) assert ("token",) in [tuple(e["loc"]) for e in exc.value.errors()] def test_output_models_still_build_with_declared_fields() -> None: """The forbid boundary rejects the credential, not the row: the declared-field construction every endpoint uses still validates.""" out = GitSourceOut(**_out_kwargs()) assert out.url == URL row = GitSourceRow(**_row_kwargs()) assert row.url == URL and row.kind == "git" # --------------------------------------------------------------------------- # sanitize_url (task 02) — the OUTPUT mask: userinfo stripped from # https? URLs only, byte-identical for everything else # --------------------------------------------------------------------------- def test_sanitize_strips_user_pass_userinfo() -> None: """The TODO L5 shape — the embedded credential is gone, the bare host/path survive character for character.""" assert ( sanitize_url("https://myuser:ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx@github.com/owner/private-repo.git") == "https://github.com/owner/private-repo.git" ) def test_sanitize_strips_username_as_token_form() -> None: """The documented GitHub shape (no colon in the userinfo): the whole run is stripped too.""" assert sanitize_url("https://ghp_onlytoken@github.com/owner/repo.git") == ( "https://github.com/owner/repo.git" ) def test_sanitize_handles_http_scheme() -> None: assert sanitize_url("http://user:pass@git.local/repo.git") == "http://git.local/repo.git" def test_sanitize_leaves_ssh_scp_and_local_untouched() -> None: """``ssh://``, scp-style ``git@``, and local paths carry no userinfo (or are not secrets in this shape) — untouched.""" for url in ( "ssh://git@example.com/repo.git", "git@github.com:owner/repo.git", "/home/owner/bor-sources/repo", "~/Homelab", ): assert sanitize_url(url) == url, url def test_sanitize_is_byte_identical_for_credential_free_urls() -> None: """The phase-50/35 contract: stored URLs surface verbatim when they carry no credential — an anchored regex replace, never a URL parser re-serialization (a parser would rewrite the path).""" url = "https://github.com/owner/repo.git?ref=main#readme" assert sanitize_url(url) == url def test_sanitize_at_in_path_is_not_userinfo() -> None: """A ``@`` inside the *path* is not userinfo — the anchor requires the run to come right after the scheme.""" url = "https://github.com/owner/repo@v1/blob/x" assert sanitize_url(url) == url def test_sanitize_is_idempotent() -> None: embedded = "https://user:pass@github.com/owner/repo.git" once = sanitize_url(embedded) assert sanitize_url(once) == once def test_sanitize_empty_and_scheme_only_urls() -> None: assert sanitize_url("") == "" assert sanitize_url("https://") == "https://" # --------------------------------------------------------------------------- # clone_url_for (task 02) — the CLONE-time credential injection # --------------------------------------------------------------------------- def test_clone_url_null_token_returns_stored_url_verbatim() -> None: """Public repos (and local rows) clone byte-identically to pre-phase-121 — no injection, no logging.""" row = GitSource(url="https://github.com/owner/public.git", token=None) assert clone_url_for(row) == "https://github.com/owner/public.git" def test_clone_url_empty_token_is_falsy_no_injection() -> None: row = GitSource(url="https://github.com/owner/public.git", token="") assert clone_url_for(row) == "https://github.com/owner/public.git" def test_clone_url_injects_x_access_token_for_https() -> None: """The task-02 assumption (step 4): ``x-access-token`` as the userinfo username — GitHub-agnostic, reads as non-identifying.""" row = GitSource(url="https://github.com/owner/private.git", token="ghp_secret123") assert ( clone_url_for(row) == "https://x-access-token:ghp_secret123@github.com/owner/private.git" ) def test_clone_url_injects_for_http_scheme() -> None: row = GitSource(url="http://git.local/repo.git", token="tok") assert clone_url_for(row) == "http://x-access-token:tok@git.local/repo.git" def test_clone_url_replaces_existing_userinfo_with_column_credential() -> None: """A legacy stored URL that still embeds userinfo (only reachable via direct-DB writes now — the API normalizes at write time) gets the column credential, not the embedded one.""" row = GitSource(url="https://user:oldpass@github.com/owner/repo.git", token="newpass") assert clone_url_for(row) == "https://x-access-token:newpass@github.com/owner/repo.git" def test_clone_url_legacy_row_clones_with_original_stored_url() -> None: """Completion criterion: a pre-phase row (credential embedded in the stored URL, token NULL) produces the ORIGINAL stored URL at clone time — the credential keeps working.""" stored = "https://user:ghp_secret@github.com/owner/repo.git" row = GitSource(url=stored, token=None) assert clone_url_for(row) == stored @pytest.mark.parametrize( ("url", "token"), [ ("ssh://git@example.com/repo.git", "sekrit-value"), ("git@github.com:owner/repo.git", "sekrit-value"), ("/home/owner/bor-sources/repo", "sekrit-value"), ], ) def test_clone_url_non_https_token_is_noop_with_warning( caplog: pytest.LogCaptureFixture, url: str, token: str ) -> None: """A token cannot authenticate ssh/scp/local — the URL is returned unchanged, a warning is logged (no crash), and the token VALUE never leaks into the log line.""" row = GitSource(url=url, token=token) with caplog.at_level("WARNING", logger="app.rag.git_sources"): assert clone_url_for(row) == url warnings = [r for r in caplog.records if r.levelname == "WARNING"] assert len(warnings) == 1 assert "sekrit-value" not in warnings[0].getMessage() # --------------------------------------------------------------------------- # normalize_credential (task 02) — the WRITE-path normalizer # --------------------------------------------------------------------------- def test_normalize_moves_embedded_password_to_column() -> None: """The TODO L5 paste: ``user:pass@`` → bare URL + the PASSWORD part as the token (the username is an identifier, not the credential — the clone-time injection ``x-access-token:@`` must authenticate).""" bare, effective = normalize_credential( "https://myuser:ghp_secret123@github.com/owner/private-repo.git", None ) assert bare == "https://github.com/owner/private-repo.git" assert effective == "ghp_secret123" def test_normalize_password_may_contain_colons() -> None: """The split is on the FIRST colon only — ``user:a:b`` moves ``a:b`` wholesale.""" bare, effective = normalize_credential("https://user:a:b@github.com/x/y.git", None) assert bare == "https://github.com/x/y.git" assert effective == "a:b" def test_normalize_username_as_token_form_moves_whole_run() -> None: """The username-as-token form (no colon in the userinfo) — the whole run is the credential.""" bare, effective = normalize_credential("https://ghp_onlytoken@github.com/x/y.git", None) assert bare == "https://github.com/x/y.git" assert effective == "ghp_onlytoken" def test_normalize_explicit_token_wins_over_embedded() -> None: """LOCKED A6 — explicit beats embedded (the masked field and the pasted URL disagree: the field is the intent).""" bare, effective = normalize_credential( "https://user:ghp_embedded@github.com/x/y.git", "ghp_explicit" ) assert bare == "https://github.com/x/y.git" assert effective == "ghp_explicit" def test_normalize_explicit_empty_token_wins_and_clears() -> None: """An explicit "" (blank masked field) is a deliberate no-credential declaration: it beats the embedded one and stores NULL (the caller stores ``effective or None``).""" bare, effective = normalize_credential("https://user:ghp_x@github.com/x/y.git", "") assert bare == "https://github.com/x/y.git" assert effective == "" def test_normalize_clean_url_with_explicit_token_untouched() -> None: """The common UI path (bare URL + masked field): the URL is byte-identical, the token passes through.""" bare, effective = normalize_credential("https://github.com/x/y.git", "ghp_t") assert bare == "https://github.com/x/y.git" assert effective == "ghp_t" def test_normalize_clean_url_without_token_untouched() -> None: assert normalize_credential("https://github.com/x/y.git", None) == ( "https://github.com/x/y.git", None, ) def test_normalize_non_https_urls_untouched() -> None: """ssh/scp/local carry no userinfo — returned untouched, token (whatever it is) passing through for the caller's tri-state.""" for url in ("ssh://git@example.com/repo.git", "git@github.com:x/y.git", "/local/dir"): assert normalize_credential(url, None) == (url, None) assert normalize_credential(url, "t") == (url, "t") def test_normalize_idempotent_on_bare_url() -> None: bare, _ = normalize_credential("https://user:pass@github.com/x/y.git", None) again, effective = normalize_credential(bare, "tok") assert again == bare assert effective == "tok" # --------------------------------------------------------------------------- # Frontend source pins (task 03) — the masked token field: add form + # per-row editor; the display sites stay on the server-sanitized s.url # --------------------------------------------------------------------------- FRONTEND = Path(__file__).resolve().parents[2] / "frontend" SHELL_HTML = FRONTEND / "index.html" GIT_SOURCES_JS = FRONTEND / "assets" / "git-sources.js" STYLES_CSS = FRONTEND / "assets" / "styles.css" ADD_TOKEN_PLACEHOLDER = "ghp_… or another PAT" EDIT_TOKEN_PLACEHOLDER = "leave blank to keep the current token" ADD_BODY_EXPR = "(url, token) => ({ url, ...(token ? { token } : {}) })" PATCH_BODY_EXPR = "JSON.stringify({ ignore_paths: lines, ...(token ? { token } : {}) })" def _read(path: Path) -> str: return path.read_text(encoding="utf-8") def _js_text() -> str: return _read(GIT_SOURCES_JS) def _fn(js: str, name: str) -> str: """The source of a (possibly async) top-level function via balanced-brace counting (the test_source_ignore_paths.py helper).""" for prefix in ("async function ", "function "): start = js.find(f"{prefix}{name}(") if start != -1: depth = 0 for i in range(js.find("{", start), len(js)): if js[i] == "{": depth += 1 elif js[i] == "}": depth -= 1 if depth == 0: return js[start : i + 1] raise AssertionError(f"unbalanced braces in {name}()") raise AssertionError(f"{name}() must exist in git-sources.js") def _input_tag(html: str, id_attr: str) -> str: m = re.search(rf"]*id=\"{id_attr}\"[^>]*>", html, re.S) assert m, f"missing in the shell" return m.group(0) def test_add_form_token_field_is_masked_and_optional() -> None: """The add form's `#git-source-token` (task 03 step 1, LOCKED A2): ``type="password"`` (masked — the credential is never a visible-text field in the DOM), ``autocomplete="off"`` (a PAT is not a site credential — no browser save offer), the 500-char cap mirrored, the placeholder, INSIDE `#git-source-form` before the submit button, and a visible `", html, re.S) assert label, "a visible ", html, re.S) assert label and "Token" in label.group(1), "a visible label (WCAG)" js = _js_text() open_body = _fn(js, "openIgnoreEditor") close_body = _fn(js, "closeIgnoreEditor") assert 'ignoreTokenInput.value = ""' in open_body, ("opens BLANK — never prefilled") assert 'ignoreTokenInput.value = ""' in close_body, ("resets on close") def test_editor_patch_omits_a_blank_token() -> None: """saveIgnorePaths: the token is read AFTER the line parse and the PATCH body is `{ ignore_paths: lines, ...(token ? { token } : {}) }` — a BLANK token omits the key (the task-01 tri-state: absent = no change, the row's stored credential is kept).""" js = _js_text() save = _fn(js, "saveIgnorePaths") drop_i = save.find(".filter(Boolean)") read_i = save.find('ignoreTokenInput ? ignoreTokenInput.value.trim() : ""', drop_i) body_i = save.find(PATCH_BODY_EXPR, read_i) assert -1 < drop_i < read_i < body_i, ("blank token → key omitted (tri-state no-change)") def test_display_sites_render_the_server_bare_url_only() -> None: """Task 03 step 3: every display site keeps rendering `s.url` UNCHANGED (the server now returns bare URLs — sanitize_url, phase 121), the one-line source note pins that contract, and NO display site ever reads a token off the row (the response has no token field — and even if one did, the UI must never re-embed it).""" js = _js_text() make = _fn(js, "makeRow") assert "const value = isLocal ? (s.path ?? s.url) : s.url;" in make, ( "the display value is the server row's url/path — UNCHANGED" ) assert "sanitized server-side" in make, ("the one-line phase-121 note at the display site") assert "never re-embed a credential" in make assert "s.token" not in js, "the UI never re-embeds a credential (LOCKED A2)" def test_module_docstring_carries_the_phase_121_token_contract() -> None: """The git-sources.js module docstring gained the phase-121 bullet: the masked add field (type=password, autocomplete=off, the "optional — private repos" hint), the blank-omits-key submit body, the editor's "leave blank to keep the current token" mirror, and the display-sites-unchanged (sanitize_url) contract.""" doc = _js_text().split("*/", 2)[0] # the module docstring (first block) for frag in ( "Phase 121 (task 03)", "#git-source-token", "type=password", "optional — private repos", ADD_BODY_EXPR, "leave blank to keep the current token", "sanitize_url", ): assert frag in doc, f"the module docstring lost: {frag!r}" def test_token_field_css_rules_are_house_styled() -> None: """styles.css (task 03 step 2 — no new CSS beyond the theme's input treatment): the add form's token input shares the URL input's box (grouped selector — mono, >=44px floor); the `.field-hint` rule is muted (ink-soft — 5.1:1 on the label's surface, AA) and small; the editor's token field is full panel width at the 44px floor; the mobile media query groups it in the same min-width:0 rule as the URL input.""" css = _read(STYLES_CSS) grouped = css.find("#git-source-url,\n#git-source-token {") assert grouped != -1, "#git-source-token shares #git-source-url's box" rule = css[grouped : css.find("}", grouped)] assert "min-height: 44px" in rule and "var(--mono)" in rule hint = css.find(".field-hint {") assert hint != -1, "the .field-hint rule" hint_rule = css[hint : css.find("}", hint)] assert "var(--ink-soft)" in hint_rule and "font-size: 0.8rem" in hint_rule tok = css.find(".ignore-editor-token {") assert tok != -1, "the editor's token field rule" tok_rule = css[tok : css.find("}", tok)] assert "width: 100%" in tok_rule and "min-height: 44px" in tok_rule mobile = css.find("#git-source-url,\n #git-source-token,") assert mobile != -1, "the mobile squeeze groups the token input"