@samitouri / QOSami_Agents_Zero-2
Code Commits Issues Pulls Sessions CI/CD
main
main
py 160 lines 5.22 KB
Raw
1 from __future__ import annotations
2
3 import stat
4 import sys
5 from pathlib import Path
6
7
8 PROJECT_ROOT = Path(__file__).resolve().parents[1]
9 if str(PROJECT_ROOT) not in sys.path:
10 sys.path.insert(0, str(PROJECT_ROOT))
11
12 from extensions.python.startup_migration import _10_self_update_manager as migration
13
14
15 SAFE_SOURCE = """#!/usr/bin/env python3
16 from pathlib import Path
17
18
19 REPO_DIR = Path("/a0")
20
21
22 def should_include_usr_backup_entry(source_file, logger):
23 logger.log("Skipping non-regular usr backup entry")
24 return False
25
26
27 def clean_transient_desktop_agent_state(repo_dir, logger):
28 return None
29
30
31 def refresh_codex_cli(logger):
32 return None
33
34
35 def docker_run_ui():
36 clean_transient_desktop_agent_state(REPO_DIR, logger)
37 refresh_codex_cli(logger)
38 """
39
40
41 def test_self_update_runtime_sync_replaces_stale_manager(tmp_path):
42 source = tmp_path / "source_self_update_manager.py"
43 target = tmp_path / "self_update_manager.py"
44 stale = "# old updater without non-regular usr backup guards\n"
45 source.write_text(SAFE_SOURCE, encoding="utf-8")
46 target.write_text(stale, encoding="utf-8")
47 target.chmod(0o600)
48
49 result = migration.ensure_self_update_manager_runtime_current(
50 target_path=target,
51 source_path=source,
52 )
53
54 assert result["ok"] is True
55 assert result["updated"] is True
56 assert target.read_text(encoding="utf-8") == SAFE_SOURCE
57 assert (target.stat().st_mode & 0o777) == 0o600
58 backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}")
59 assert backup.read_text(encoding="utf-8") == stale
60
61
62 def test_self_update_runtime_sync_accepts_repository_manager_source(tmp_path):
63 source = PROJECT_ROOT / "docker" / "run" / "fs" / "exe" / "self_update_manager.py"
64 target = tmp_path / "self_update_manager.py"
65 stale = "# old updater without non-regular usr backup guards\n"
66 target.write_text(stale, encoding="utf-8")
67
68 result = migration.ensure_self_update_manager_runtime_current(
69 target_path=target,
70 source_path=source,
71 )
72
73 assert result["ok"] is True
74 assert result["updated"] is True
75 assert target.read_text(encoding="utf-8") == source.read_text(encoding="utf-8")
76
77
78 def test_self_update_runtime_sync_starts_codex_refresh(monkeypatch, tmp_path):
79 calls = []
80 manager_path = tmp_path / "self_update_manager.py"
81 monkeypatch.setattr(
82 migration.subprocess,
83 "Popen",
84 lambda *args, **kwargs: calls.append((args, kwargs)),
85 )
86
87 assert migration.start_codex_cli_refresh(manager_path) == ""
88 assert calls[0][0][0] == [sys.executable, str(manager_path), "refresh-codex"]
89 assert calls[0][1]["start_new_session"] is True
90
91
92 def test_self_update_runtime_sync_skips_current_manager(tmp_path):
93 source = tmp_path / "source_self_update_manager.py"
94 target = tmp_path / "self_update_manager.py"
95 source.write_text(SAFE_SOURCE, encoding="utf-8")
96 target.write_text(SAFE_SOURCE, encoding="utf-8")
97
98 result = migration.ensure_self_update_manager_runtime_current(
99 target_path=target,
100 source_path=source,
101 )
102
103 assert result == {"ok": True, "updated": False, "reason": "already-current"}
104 backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}")
105 assert not backup.exists()
106
107
108 def test_self_update_runtime_sync_refuses_source_without_required_markers(tmp_path):
109 source = tmp_path / "source_self_update_manager.py"
110 target = tmp_path / "self_update_manager.py"
111 stale = "# old updater without non-regular usr backup guards\n"
112 source.write_text("def create_usr_backup():\n pass\n", encoding="utf-8")
113 target.write_text(stale, encoding="utf-8")
114
115 result = migration.ensure_self_update_manager_runtime_current(
116 target_path=target,
117 source_path=source,
118 )
119
120 assert result["ok"] is False
121 assert result["updated"] is False
122 assert "missing required safety markers" in result["warning"]
123 assert target.read_text(encoding="utf-8") == stale
124 backup = target.with_name(f"{target.name}{migration.BACKUP_SUFFIX}")
125 assert not backup.exists()
126
127
128 def test_self_update_runtime_sync_missing_target_is_quiet(tmp_path):
129 source = tmp_path / "source_self_update_manager.py"
130 target = tmp_path / "missing_self_update_manager.py"
131 source.write_text(SAFE_SOURCE, encoding="utf-8")
132
133 result = migration.ensure_self_update_manager_runtime_current(
134 target_path=target,
135 source_path=source,
136 )
137
138 assert result["ok"] is True
139 assert result["updated"] is False
140 assert "not found" in result["reason"]
141
142
143 def test_self_update_runtime_sync_skips_non_regular_target(tmp_path):
144 source = tmp_path / "source_self_update_manager.py"
145 target = tmp_path / "self_update_manager.py"
146 link_target = tmp_path / "linked_self_update_manager.py"
147 source.write_text(SAFE_SOURCE, encoding="utf-8")
148 link_target.write_text("# linked updater\n", encoding="utf-8")
149 target.symlink_to(link_target)
150
151 result = migration.ensure_self_update_manager_runtime_current(
152 target_path=target,
153 source_path=source,
154 )
155
156 assert result["ok"] is True
157 assert result["updated"] is False
158 assert "not a regular file" in result["reason"]
159 assert stat.S_ISLNK(target.lstat().st_mode)
160 assert link_target.read_text(encoding="utf-8") == "# linked updater\n"