from __future__ import annotations import importlib.util import json import os import subprocess import tempfile import unittest from types import SimpleNamespace from pathlib import Path from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] import sys sys.path.insert(0, str(ROOT / "runtime" / "agent-os")) sys.path.insert(0, str(ROOT / "active-runtime" / "app")) from gitea_tag_publication import build_publisher import authoritative_git as authoritative_git_module from authoritative_git import ( AuthoritativeGitError, AuthoritativeGitNotFound, AuthoritativeGitOutputTooLarge, authoritative_branches, authoritative_diff, authoritative_commit, authoritative_head_oid, authoritative_tags, repository_snapshot, run_authoritative_git, validate_git_revision, verify_repository, ) API_SPEC = importlib.util.spec_from_file_location( "authoritative_test_api", ROOT / "active-runtime" / "app" / "aios_chatgpt_actions_api.py", ) assert API_SPEC and API_SPEC.loader api = importlib.util.module_from_spec(API_SPEC) API_SPEC.loader.exec_module(api) class AuthoritativeGitContextTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.bare = self.root / "repo.git" self.seed = self.root / "seed" self.runtime = self.root / "runtime" subprocess.run(["git", "init", "--bare", str(self.bare)], check=True, capture_output=True) subprocess.run(["git", "init", "-b", "main", str(self.seed)], check=True, capture_output=True) for key, value in (("user.name", "test"), ("user.email", "test@example.invalid")): subprocess.run(["git", "-C", str(self.seed), "config", key, value], check=True) (self.seed / "README").write_text("context\n", encoding="utf-8") subprocess.run(["git", "-C", str(self.seed), "add", "README"], check=True) subprocess.run(["git", "-C", str(self.seed), "commit", "-m", "seed"], check=True, capture_output=True) subprocess.run(["git", "-C", str(self.seed), "remote", "add", "origin", str(self.bare)], check=True) subprocess.run(["git", "-C", str(self.seed), "push", "origin", "main"], check=True, capture_output=True) subprocess.run(["git", "--git-dir", str(self.bare), "config", "remote.gitea.url", "https://g.cnnetmkt.top/git/aios/aios.git"], check=True) subprocess.run(["git", "-C", str(self.seed), "tag", "-a", "V1R1C0", "-m", "annotated"], check=True) subprocess.run(["git", "-C", str(self.seed), "tag", "V1R1C1"], check=True) subprocess.run(["git", "-C", str(self.seed), "checkout", "-b", "patch/context"], check=True, capture_output=True) (self.seed / "CHANGE").write_text("authoritative diff\n", encoding="utf-8") subprocess.run(["git", "-C", str(self.seed), "add", "CHANGE"], check=True) subprocess.run(["git", "-C", str(self.seed), "commit", "-m", "change"], check=True, capture_output=True) subprocess.run(["git", "-C", str(self.seed), "push", "origin", "patch/context", "--tags"], check=True, capture_output=True) (self.runtime / "versions").mkdir(parents=True) (self.runtime / "patches").mkdir() (self.runtime / "versions" / "V1R1C0.json").write_text( json.dumps({"version": "V1R1C0"}), encoding="utf-8" ) (self.runtime / "patches" / "context.json").write_text( json.dumps({"branch": "patch/context"}), encoding="utf-8" ) self.env = { "AIOS_GITOPS_AUTHORITATIVE_GIT_DIR": str(self.bare), "AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios", } def tearDown(self): self.tmp.cleanup() def test_runtime_cwd_without_git_uses_authoritative_head(self): with patch.dict(os.environ, self.env, clear=False), patch("os.getcwd", return_value=str(self.runtime)): self.assertFalse((self.runtime / ".git").exists()) self.assertEqual(len(authoritative_head_oid()), 40) def test_annotated_and_lightweight_tags_include_type_and_peeled_target(self): with patch.dict(os.environ, self.env, clear=False): tags = {item["name"]: item for item in authoritative_tags()} self.assertEqual(tags["V1R1C0"]["object_type"], "tag") self.assertEqual(tags["V1R1C0"]["peeled_target_oid"], tags["V1R1C1"]["peeled_target_oid"]) self.assertEqual(tags["V1R1C1"]["object_type"], "commit") def test_authoritative_branches_include_exact_oids(self): with patch.dict(os.environ, self.env, clear=False): branches = {item["name"]: item["oid"] for item in authoritative_branches()} self.assertIn("main", branches) self.assertIn("patch/context", branches) self.assertEqual(len(branches["main"]), 40) def test_runtime_cwd_changes_do_not_change_results(self): old_cwd = os.getcwd() try: with patch.dict(os.environ, self.env, clear=False): first = authoritative_head_oid() os.chdir(str(self.runtime)) second = authoritative_head_oid() self.assertEqual(first, second) finally: os.chdir(old_cwd) def test_missing_configuration_fails_closed(self): with patch.dict(os.environ, {}, clear=True), self.assertRaises(AuthoritativeGitError): verify_repository() def test_missing_repository_fails_closed(self): with patch.dict(os.environ, {**self.env, "AIOS_GITOPS_AUTHORITATIVE_GIT_DIR": str(self.root / "missing")}, clear=False): with self.assertRaises(AuthoritativeGitError): verify_repository() def test_non_bare_repository_fails_closed(self): with patch.dict(os.environ, self.env, clear=False), self.assertRaises(AuthoritativeGitError): verify_repository(self.seed) def test_repository_identity_mismatch_fails_closed(self): with patch.dict(os.environ, {**self.env, "AIOS_GITOPS_REPOSITORY_IDENTITY": "other/repo"}, clear=False): with self.assertRaises(AuthoritativeGitError): verify_repository() def test_git_command_failure_is_not_an_empty_success(self): with patch.dict(os.environ, self.env, clear=False): with self.assertRaises(AuthoritativeGitError) as ctx: run_authoritative_git(["rev-parse", "--verify", "refs/heads/no-such-branch"]) self.assertIn("GIT_REF_QUERY_FAILED", str(ctx.exception)) self.assertNotIn("tags=[]", str(ctx.exception)) def test_commit_input_rejects_git_parameter_injection(self): with patch.dict(os.environ, self.env, clear=False): with self.assertRaises(ValueError): authoritative_commit("1" * 39 + ";") def test_repository_snapshot_exposes_authoritative_head(self): with patch.dict(os.environ, self.env, clear=False): snapshot = repository_snapshot() self.assertTrue(snapshot["repository_available"]) self.assertEqual(snapshot["repository_identity"], "aios/aios") self.assertEqual(len(snapshot["authoritative_head_oid"]), 40) def _get(self, path): handler = object.__new__(api.H) handler.path = path handler.sendj = Mock(side_effect=lambda data, code=200: (data, code)) handler.sendt = Mock(side_effect=lambda data, ctype="text/plain; charset=utf-8", code=200: (data, code)) with ( patch.dict(os.environ, self.env, clear=False), patch.object(api, "WS", self.runtime), patch.object(api, "AUTHORITATIVE_GIT_DIR", self.bare), patch.object(api, "REPOSITORY_IDENTITY", "aios/aios"), ): api.H.do_GET(handler) args = handler.sendj.call_args.args return args[0], args[1] if len(args) > 1 else 200 def test_api_list_versions_uses_authoritative_tags(self): body, code = self._get("/api/ver") self.assertEqual(code, 200) self.assertTrue(body["repository_available"]) self.assertTrue(body["local_authoritative_tags"]) self.assertEqual(body["head"], body["authoritative_head_oid"]) self.assertIn("V1R1C0", body["tags"]) def test_api_list_patches_uses_authoritative_branches(self): body, code = self._get("/api/patch") self.assertEqual(code, 200) self.assertIn("patch/context", body["branches"]) self.assertTrue(body["authoritative_branches"]) def test_api_unavailable_does_not_return_empty_lists(self): handler = object.__new__(api.H) handler.path = "/api/ver" handler.sendj = Mock(side_effect=lambda data, code=200: (data, code)) with patch.dict(os.environ, {}, clear=True), patch.object(api, "WS", self.runtime), patch.object(api, "AUTHORITATIVE_GIT_DIR", None), patch.object(api, "REPOSITORY_IDENTITY", ""): api.H.do_GET(handler) body, code = handler.sendj.call_args.args self.assertEqual(code, 503) self.assertFalse(body["repository_available"]) self.assertIn("query_error", body) def test_api_runtime_status_contains_release_and_repository_fields(self): handler = object.__new__(api.H) handler.path = "/api/run" handler.sendj = Mock(side_effect=lambda data, code=200: (data, code)) real_run = subprocess.run def service_or_real(*args, **kwargs): command = args[0] if args else kwargs.get("args", []) if command and command[0] == "systemctl": return subprocess.CompletedProcess(command, 0, stdout="active\n", stderr="") return real_run(*args, **kwargs) with ( patch.dict(os.environ, self.env, clear=False), patch.object(api, "WS", self.runtime), patch.object(api, "AUTHORITATIVE_GIT_DIR", self.bare), patch.object(api, "REPOSITORY_IDENTITY", "aios/aios"), patch.object(api.subprocess, "run", side_effect=service_or_real), ): api.H.do_GET(handler) body, code = handler.sendj.call_args.args self.assertEqual(code, 200) self.assertIsNone(body["workspace_branch"]) self.assertTrue(body["authoritative_git_dir_configured"]) self.assertEqual(body["authoritative_head_oid"], body["authoritative_head_oid"]) def test_authoritative_head_is_independent_of_runtime_directory(self): with patch.dict(os.environ, self.env, clear=False): old = os.getcwd() try: os.chdir(str(self.runtime)) self.assertEqual(authoritative_head_oid(), authoritative_head_oid()) finally: os.chdir(old) def test_authoritative_api_is_read_only(self): before = subprocess.run(["git", "--git-dir", str(self.bare), "show-ref"], check=True, capture_output=True, text=True).stdout with patch.dict(os.environ, self.env, clear=False): authoritative_tags() authoritative_branches() after = subprocess.run(["git", "--git-dir", str(self.bare), "show-ref"], check=True, capture_output=True, text=True).stdout self.assertEqual(before, after) def test_error_text_does_not_include_credential_values(self): with patch.dict(os.environ, self.env, clear=False): with self.assertRaises(AuthoritativeGitError) as ctx: run_authoritative_git(["rev-parse", "--verify", "https://user:password@example.invalid/x"]) self.assertNotIn("password", str(ctx.exception).lower()) def test_cli_publisher_requires_and_uses_authoritative_repository(self): args = SimpleNamespace( workspace=str(self.runtime), remote="gitea", gitea_api="http://127.0.0.1:3000/api/v1", gitea_repo="aios/aios", gitea_token_file="/dev/null", ) with patch.dict(os.environ, self.env, clear=False): publisher = build_publisher(args) self.assertEqual(publisher.git.authoritative_repo, self.bare) def test_cli_publisher_fails_closed_without_authoritative_repository(self): args = SimpleNamespace( workspace=str(self.runtime), remote="gitea", gitea_api="http://127.0.0.1:3000/api/v1", gitea_repo="aios/aios", gitea_token_file="/dev/null", ) with patch.dict(os.environ, {}, clear=True), self.assertRaises(RuntimeError): build_publisher(args) def test_authoritative_diff_supports_branch_tag_and_exact_oid(self): with patch.dict(os.environ, self.env, clear=True): by_ref = authoritative_diff("V1R1C0", "patch/context") main_oid = validate_git_revision("refs/heads/main", "from") patch_oid = validate_git_revision("refs/heads/patch/context", "to") by_oid = authoritative_diff(main_oid, patch_oid) self.assertIn("CHANGE", by_ref["stdout"]) self.assertEqual(by_ref["stdout"], by_oid["stdout"]) self.assertEqual(by_ref["code"], 0) def test_authoritative_diff_is_cwd_independent(self): old = os.getcwd() try: with patch.dict(os.environ, self.env, clear=True): first = authoritative_diff("V1R1C0", "patch/context")["stdout"] os.chdir(self.runtime) second = authoritative_diff("V1R1C0", "patch/context")["stdout"] self.assertEqual(first, second) finally: os.chdir(old) def test_revision_injection_and_revision_expressions_are_rejected(self): with patch.dict(os.environ, self.env, clear=True): for value in ("--help", "--output=x", "HEAD~1", "main^{tree}", "@{1}", ":/secret", "", " main"): with self.subTest(value=value), self.assertRaises(ValueError): validate_git_revision(value, "from") def test_valid_but_missing_revision_is_not_found(self): with patch.dict(os.environ, self.env, clear=True), self.assertRaises(AuthoritativeGitNotFound): validate_git_revision("missing-branch", "from") def test_diff_output_limit_is_fail_closed(self): with patch.dict(os.environ, self.env, clear=True), patch.object(authoritative_git_module, "_MAX_OUTPUT", 1): with self.assertRaises(AuthoritativeGitOutputTooLarge): authoritative_diff("V1R1C0", "patch/context") def test_api_diff_uses_authoritative_repository(self): body, code = self._get("/api/diff?from=V1R1C0&to=patch/context") self.assertEqual(code, 200) self.assertIn("CHANGE", body["diff"]["stdout"]) self.assertEqual(len(body["diff"]["from_oid"]), 40) def test_api_diff_status_mapping(self): _, code = self._get("/api/diff?from=--help&to=patch/context") self.assertEqual(code, 400) _, code = self._get("/api/diff?from=missing-branch&to=patch/context") self.assertEqual(code, 404) with patch.object(api, "authoritative_diff", side_effect=AuthoritativeGitOutputTooLarge("GIT_OUTPUT_LIMIT_EXCEEDED")): _, code = self._get("/api/diff?from=V1R1C0&to=patch/context") self.assertEqual(code, 413) def test_http_publisher_fails_closed_without_authoritative_repository(self): with patch.dict(os.environ, {}, clear=True), patch.object(api, "AUTHORITATIVE_GIT_DIR", None), patch.object(api, "REPOSITORY_IDENTITY", ""): with self.assertRaises(RuntimeError): api.publication_publisher() def test_http_publisher_fails_closed_on_identity_mismatch(self): with patch.dict(os.environ, self.env, clear=True), patch.object(api, "AUTHORITATIVE_GIT_DIR", self.bare), patch.object(api, "REPOSITORY_IDENTITY", "other/repo"): with self.assertRaises(RuntimeError): api.publication_publisher() def test_writer_guard_remains_false(self): decision = api.evaluate_git_write(self.runtime, {"AIOS_GITOPS_WRITES_ENABLED": "false"}) self.assertFalse(decision.allowed) self.assertEqual(decision.status_code, 423)