244 lines
10 KiB
Python
244 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from cleanup_project import apply_plan, plan_cleanup
|
|
from hockey_data.client import Stat2TVClient
|
|
from hockey_data.snapshot_storage import save_xml_snapshot, trim_discovery_reports
|
|
from ui_builder.manager import DEFAULT_CONFIG, UIBuilderManager
|
|
from ui_builder.storage import prepare_ui_storage
|
|
|
|
|
|
def test_identical_xml_reuses_path_and_original_timestamp(tmp_path):
|
|
first = save_xml_snapshot(tmp_path, b"<items/>", resource="tournaments")
|
|
os.utime(first, (1, 1))
|
|
second = save_xml_snapshot(tmp_path, b"<items/>", resource="tournaments")
|
|
assert first == second and second.stat().st_mtime == 1
|
|
assert len(list(tmp_path.glob("*.xml"))) == 1
|
|
assert save_xml_snapshot(tmp_path, b"<changed/>", resource="tournaments") != first
|
|
outside = save_xml_snapshot(tmp_path, b"x", resource="../../data", tournament="../../outside")
|
|
assert outside.parent == tmp_path and outside.read_bytes() == b"x"
|
|
|
|
|
|
def test_report_retention_preserves_xml_other_json_and_recent_reports(tmp_path):
|
|
old = time.time() - 90 * 86400
|
|
for i in range(45):
|
|
path = tmp_path / f"20260101_000000_{i:06d}_1436_games_discovery.json"
|
|
path.write_text("{}")
|
|
os.utime(path, (old, old))
|
|
xml = tmp_path / "20260101_000000_000001_1436_games.xml"
|
|
xml.write_text("<games/>")
|
|
config = tmp_path / "credentials.json"
|
|
config.write_text("{}")
|
|
assert trim_discovery_reports(tmp_path) == 5
|
|
assert xml.exists() and config.exists()
|
|
assert len(list(tmp_path.glob("*_games_discovery.json"))) == 40
|
|
|
|
|
|
def test_failed_snapshot_replacement_leaves_existing_bytes_and_no_temp(tmp_path, monkeypatch):
|
|
path = save_xml_snapshot(tmp_path, b"<items/>", resource="games")
|
|
path.write_bytes(b"previous damaged diagnostic")
|
|
def fail(*args):
|
|
raise OSError("simulated failure")
|
|
monkeypatch.setattr("hockey_data.snapshot_storage.os.replace", fail)
|
|
with pytest.raises(OSError):
|
|
save_xml_snapshot(tmp_path, b"<items/>", resource="games")
|
|
assert path.read_bytes() == b"previous damaged diagnostic"
|
|
assert not list(tmp_path.glob("*.tmp"))
|
|
|
|
|
|
def test_noop_ui_save_does_not_consume_backup_history(tmp_path):
|
|
manager = UIBuilderManager(tmp_path)
|
|
original = manager.settings_file.stat().st_mtime_ns
|
|
manager.save(manager.load())
|
|
assert not manager.list_backups()
|
|
assert manager.settings_file.stat().st_mtime_ns == original
|
|
config = manager.load()
|
|
config["project_name"] = "Changed"
|
|
manager.save(config)
|
|
assert len(manager.list_backups()) == 1
|
|
|
|
|
|
def test_backup_migration_does_not_resurrect_trimmed_copies(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("HOCKEY_UI_DATA_DIR", raising=False)
|
|
root = tmp_path / "settings"
|
|
root.mkdir()
|
|
(root / "ui_builder_draft.json").write_text(json.dumps(DEFAULT_CONFIG))
|
|
(root / "backups").mkdir()
|
|
name = "ui_builder_draft_20260101_000000_000000.json"
|
|
(root / "backups" / name).write_text("{}")
|
|
local = prepare_ui_storage(root, filenames=("ui_builder_draft.json",), legacy_filename="ui_builder.json")
|
|
copied = local / "backups" / name
|
|
assert copied.exists()
|
|
copied.unlink()
|
|
prepare_ui_storage(root, filenames=("ui_builder_draft.json",), legacy_filename="ui_builder.json")
|
|
assert not copied.exists() and (root / "backups" / name).exists()
|
|
|
|
|
|
def project(tmp_path):
|
|
for name in ("app.py", "hockey_data/service.py", "ui_builder/integration.py"):
|
|
path = tmp_path / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("# source")
|
|
return tmp_path
|
|
|
|
|
|
def test_cleanup_keeps_paths_live_data_and_rejects_nonallowlisted_deletions(tmp_path):
|
|
root = project(tmp_path)
|
|
snapshots = root / "hockey_data/snapshots"
|
|
snapshots.mkdir()
|
|
one = snapshots / "20260101_000000_000001_1436_ab12_games.xml"
|
|
two = snapshots / "20260101_000000_000002_1436_ab12_games.xml"
|
|
one.write_bytes(b"<items/>")
|
|
two.write_bytes(one.read_bytes())
|
|
protected = ("settings/local/ui_builder_draft.json", "settings/stat2tv_credentials.local.json",
|
|
"agent/agent_config.json", "khl_site/data/khl_players_all.json", "data/tournaments_seed.xml",
|
|
"agent/agent.log")
|
|
for name in protected:
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("keep")
|
|
pyc = root / "hockey_data/__pycache__/service.cpython-312.pyc"
|
|
pyc.parent.mkdir()
|
|
pyc.write_bytes(b"bytecode")
|
|
plan = plan_cleanup(root)
|
|
assert one.exists() and two.exists() and pyc.exists() # planning is read-only
|
|
results = apply_plan(root, plan)
|
|
assert all(item["status"] == "done" for item in results)
|
|
assert one.read_bytes() == two.read_bytes() == b"<items/>"
|
|
assert os.path.samefile(one, two) and not pyc.exists()
|
|
assert all((root / name).read_text() == "keep" for name in protected)
|
|
bad = {"action": "delete", "path": protected[0], "bytes": 4,
|
|
"sha256": hashlib.sha256(b"keep").hexdigest(), "reason": "bytecode"}
|
|
assert apply_plan(root, [bad])[0]["detail"] == "not_allowlisted"
|
|
assert (root / protected[0]).exists()
|
|
|
|
|
|
def test_cleanup_skips_symlinks_and_changed_old_ui(tmp_path):
|
|
root = project(tmp_path)
|
|
cache = root / "hockey_data/__pycache__"
|
|
cache.mkdir()
|
|
outside = tmp_path.parent / (tmp_path.name + "-outside")
|
|
outside.write_bytes(b"keep")
|
|
(cache / "external.pyc").symlink_to(outside)
|
|
old = root / "hockey_data/ui_builder/static/app.js"
|
|
old.parent.mkdir(parents=True)
|
|
old.write_text("custom changed UI")
|
|
active = root / "ui_builder/static/app.js"
|
|
active.parent.mkdir()
|
|
active.write_text("active UI")
|
|
assert not plan_cleanup(root)
|
|
assert outside.read_bytes() == b"keep" and old.exists()
|
|
|
|
|
|
def test_cleanup_removes_only_identical_migrated_old_backup(tmp_path):
|
|
root = project(tmp_path)
|
|
name = "ui_builder_draft_20260101_000000_000000.json"
|
|
old = root / "settings/backups" / name
|
|
local = root / "settings/local/backups" / name
|
|
for path in (old, local):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("{}")
|
|
assert not plan_cleanup(root)
|
|
marker = root / "settings/local/.ui_backups_migrated_v1.json"
|
|
marker.write_text('{"completed": true}')
|
|
plan = plan_cleanup(root)
|
|
assert len(plan) == 1 and plan[0]["reason"] == "migrated_backup_copy"
|
|
local.write_text('{"changed": true}')
|
|
assert apply_plan(root, plan)[0]["status"] == "skipped"
|
|
assert old.exists() and local.exists()
|
|
local.write_text("{}")
|
|
assert apply_plan(root, plan)[0]["status"] == "done"
|
|
assert not old.exists() and local.read_text() == "{}"
|
|
|
|
|
|
def test_cleanup_archived_docs_preserves_modified_current_unknown_and_path_entries(tmp_path):
|
|
root = project(tmp_path)
|
|
(root / "docs").mkdir()
|
|
entries = {"README_BUILD172_RU.md": b"original historical note",
|
|
"README_BUILD173_RU.md": b"before operator edit",
|
|
"README.md": b"current readme", "app.py": b"# source",
|
|
".env": b"private", "my-notes.md": b"personal notes",
|
|
"../README_BUILD174_RU.md": b"outside",
|
|
"docs/README_BUILD175_RU.md": b"nested",
|
|
"settings/local/README_BUILD176_RU.md": b"settings note"}
|
|
with zipfile.ZipFile(root / "docs/history.zip", "w") as archive:
|
|
for name, content in entries.items():
|
|
archive.writestr(name, content)
|
|
for name, content in entries.items():
|
|
if not name.startswith("../"):
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
modified = root / "README_BUILD173_RU.md"
|
|
modified.write_text("operator edit")
|
|
# Neither a nested archive path nor a traversal entry authorizes root deletion.
|
|
(root / "README_BUILD174_RU.md").write_bytes(b"outside")
|
|
(root / "README_BUILD175_RU.md").write_bytes(b"nested")
|
|
plan = plan_cleanup(root)
|
|
assert [step["path"] for step in plan] == ["README_BUILD172_RU.md"]
|
|
result = apply_plan(root, plan)
|
|
assert result[0]["status"] == "done"
|
|
assert not (root / "README_BUILD172_RU.md").exists()
|
|
assert modified.read_text() == "operator edit"
|
|
for name in ("README.md", "app.py", ".env", "my-notes.md", "README_BUILD174_RU.md",
|
|
"README_BUILD175_RU.md", "settings/local/README_BUILD176_RU.md"):
|
|
assert (root / name).exists()
|
|
|
|
|
|
def test_cleanup_preserves_doc_changed_after_plan_or_missing_archive(tmp_path):
|
|
root = project(tmp_path)
|
|
(root / "docs").mkdir()
|
|
doc = root / "README_BUILD172_RU.md"
|
|
doc.write_text("original")
|
|
archive_path = root / "docs/history.zip"
|
|
with zipfile.ZipFile(archive_path, "w") as archive:
|
|
archive.writestr(doc.name, "original")
|
|
plan = plan_cleanup(root)
|
|
doc.write_text("new operator note")
|
|
assert apply_plan(root, plan)[0]["status"] == "skipped"
|
|
doc.write_text("original")
|
|
archive_path.unlink()
|
|
assert apply_plan(root, plan)[0]["status"] == "skipped"
|
|
assert doc.read_text() == "original"
|
|
|
|
|
|
def test_http_pool_reused_concurrently_with_current_timeout_and_closed(monkeypatch):
|
|
class Settings:
|
|
timeout = 4
|
|
def public(self):
|
|
return {"base_url": "https://feed.invalid", "auth_mode": "basic",
|
|
"request_timeout_seconds": self.timeout, "verify_ssl": True}
|
|
def credentials(self):
|
|
return "operator", "password"
|
|
created, timeouts = [], []
|
|
original = httpx.AsyncClient
|
|
def response(request):
|
|
timeouts.append(request.extensions["timeout"]["read"])
|
|
return httpx.Response(200, content=b"<ok/>")
|
|
def factory(**kwargs):
|
|
client = original(**kwargs, transport=httpx.MockTransport(response))
|
|
created.append(client)
|
|
return client
|
|
monkeypatch.setattr("hockey_data.client.httpx.AsyncClient", factory)
|
|
async def scenario():
|
|
settings = Settings()
|
|
service = Stat2TVClient(settings)
|
|
await asyncio.gather(service.fetch_xml("a.xml"), service.fetch_xml("b.xml"))
|
|
settings.timeout = 9
|
|
await service.fetch_xml("c.xml")
|
|
assert len(created) == 1 and timeouts == [4, 4, 9]
|
|
await service.aclose()
|
|
assert created[0].is_closed and not service._clients
|
|
await service.aclose()
|
|
asyncio.run(scenario())
|