wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(security): ingest gitea backup assets

Browse Source
This commit is contained in:
ogt
2026-07-14 14:16:54 +08:00
parent 69d2e4381d
commit f963452b3e
2 changed files with 213 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
apps/api/tests/test_asset_scanner_job.py
View File

@@ -374,6 +374,9 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
async def fake_ai_catalog():
return [], []
async def fake_gitea_bundle():
return [], []
monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl)
monkeypatch.setattr(
asset_scanner_job,
@@ -390,6 +393,11 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
"_collect_ai_catalog_assets",
fake_ai_catalog,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_gitea_bundle_assets",
fake_gitea_bundle,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()
@@ -636,6 +644,11 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
"_collect_ai_catalog_assets",
empty_collector,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_gitea_bundle_assets",
empty_collector,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()
@@ -769,6 +782,71 @@ def test_ai_catalog_collector_reads_aggregates_and_model_names_only(
]
def test_gitea_bundle_readback_builds_repo_and_backup_assets() -> None:
assets, relationships = asset_scanner_job._build_gitea_bundle_assets(
{
"schema_version": "gitea_repo_bundle_backup_readback_v1",
"status": "gitea_repo_bundle_backup_readback_ready",
"host": "188",
"readback_present": True,
"ready": True,
"summary": {
"expected_repo_count": 2,
"repo_row_count": 2,
"expected_repo_missing_count": 0,
"failed_repo_count": 0,
"checksum_missing_count": 0,
"bundle_fresh": True,
"sample_restore_dry_run_ok": True,
},
"repos": [
{
"repo": "wooo/awoooi",
"present": True,
"ok": True,
"head_count": 196,
"checksum_digest_present": True,
},
{
"repo": "wooo/tsenyang-website",
"present": True,
"ok": True,
"head_count": 2,
"checksum_digest_present": True,
},
],
}
)
by_type: dict[str, list[dict]] = {}
for asset in assets:
by_type.setdefault(asset["asset_type"], []).append(asset)
assert len(by_type["gitea_repo"]) == 2
assert len(by_type["backup_target"]) == 1
assert by_type["backup_target"][0]["metadata"]["ready"] is True
assert all(
asset["metadata"]["repository_content_read_by_scanner"] is False
for asset in assets
)
assert all(
asset_scanner_job._resolve_owner_team(asset)[0] == "platform_security"
for asset in assets
)
assert relationships == [
{
"from_key": "gitea/repo/wooo/awoooi",
"to_key": "backup/gitea-bundle/188",
"relationship_type": "backs_up_to",
},
{
"from_key": "gitea/repo/wooo/tsenyang-website",
"to_key": "backup/gitea-bundle/188",
"relationship_type": "backs_up_to",
},
]
def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -> None:
terminal: dict[str, object] = {}