busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(clean-release): complete compliance redesign phases and polish tasks T047-T052

Browse Source
This commit is contained in:
busya
2026-03-10 09:11:26 +03:00
parent 030818d91f
commit b4e0e21446
79 changed files with 7430 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
backend/src/services/clean_release/repository.py
View File

@@ -9,16 +9,17 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from ...models.clean_release import (
CleanProfilePolicy,
ComplianceCheckRun,
CleanPolicySnapshot,
ComplianceRun,
ComplianceReport,
ComplianceStageRun,
ComplianceViolation,
DistributionManifest,
ReleaseCandidate,
ResourceSourceRegistry,
SourceRegistrySnapshot,
)
@@ -27,67 +28,94 @@ from ...models.clean_release import (
@dataclass
class CleanReleaseRepository:
candidates: Dict[str, ReleaseCandidate] = field(default_factory=dict)
policies: Dict[str, CleanProfilePolicy] = field(default_factory=dict)
registries: Dict[str, ResourceSourceRegistry] = field(default_factory=dict)
policies: Dict[str, CleanPolicySnapshot] = field(default_factory=dict)
registries: Dict[str, SourceRegistrySnapshot] = field(default_factory=dict)
artifacts: Dict[str, object] = field(default_factory=dict)
manifests: Dict[str, DistributionManifest] = field(default_factory=dict)
check_runs: Dict[str, ComplianceCheckRun] = field(default_factory=dict)
check_runs: Dict[str, ComplianceRun] = field(default_factory=dict)
stage_runs: Dict[str, ComplianceStageRun] = field(default_factory=dict)
reports: Dict[str, ComplianceReport] = field(default_factory=dict)
violations: Dict[str, ComplianceViolation] = field(default_factory=dict)
audit_events: List[Dict[str, Any]] = field(default_factory=list)
def save_candidate(self, candidate: ReleaseCandidate) -> ReleaseCandidate:
self.candidates[candidate.candidate_id] = candidate
self.candidates[candidate.id] = candidate
return candidate
def get_candidate(self, candidate_id: str) -> Optional[ReleaseCandidate]:
return self.candidates.get(candidate_id)
def save_policy(self, policy: CleanProfilePolicy) -> CleanProfilePolicy:
self.policies[policy.policy_id] = policy
def save_policy(self, policy: CleanPolicySnapshot) -> CleanPolicySnapshot:
self.policies[policy.id] = policy
return policy
def get_policy(self, policy_id: str) -> Optional[CleanProfilePolicy]:
def get_policy(self, policy_id: str) -> Optional[CleanPolicySnapshot]:
return self.policies.get(policy_id)
def get_active_policy(self) -> Optional[CleanProfilePolicy]:
for policy in self.policies.values():
if policy.active:
return policy
return None
def get_active_policy(self) -> Optional[CleanPolicySnapshot]:
# In-memory repo doesn't track 'active' flag on snapshot,
# this should be resolved by facade using ConfigManager.
return next(iter(self.policies.values()), None)
def save_registry(self, registry: ResourceSourceRegistry) -> ResourceSourceRegistry:
self.registries[registry.registry_id] = registry
def save_registry(self, registry: SourceRegistrySnapshot) -> SourceRegistrySnapshot:
self.registries[registry.id] = registry
return registry
def get_registry(self, registry_id: str) -> Optional[ResourceSourceRegistry]:
def get_registry(self, registry_id: str) -> Optional[SourceRegistrySnapshot]:
return self.registries.get(registry_id)
def save_artifact(self, artifact) -> object:
self.artifacts[artifact.id] = artifact
return artifact
def get_artifacts_by_candidate(self, candidate_id: str) -> List[object]:
return [a for a in self.artifacts.values() if a.candidate_id == candidate_id]
def save_manifest(self, manifest: DistributionManifest) -> DistributionManifest:
self.manifests[manifest.manifest_id] = manifest
self.manifests[manifest.id] = manifest
return manifest
def get_manifest(self, manifest_id: str) -> Optional[DistributionManifest]:
return self.manifests.get(manifest_id)
def save_check_run(self, check_run: ComplianceCheckRun) -> ComplianceCheckRun:
self.check_runs[check_run.check_run_id] = check_run
def save_distribution_manifest(self, manifest: DistributionManifest) -> DistributionManifest:
return self.save_manifest(manifest)
def get_distribution_manifest(self, manifest_id: str) -> Optional[DistributionManifest]:
return self.get_manifest(manifest_id)
def save_check_run(self, check_run: ComplianceRun) -> ComplianceRun:
self.check_runs[check_run.id] = check_run
return check_run
def get_check_run(self, check_run_id: str) -> Optional[ComplianceCheckRun]:
def get_check_run(self, check_run_id: str) -> Optional[ComplianceRun]:
return self.check_runs.get(check_run_id)
def save_compliance_run(self, run: ComplianceRun) -> ComplianceRun:
return self.save_check_run(run)
def get_compliance_run(self, run_id: str) -> Optional[ComplianceRun]:
return self.get_check_run(run_id)
def save_report(self, report: ComplianceReport) -> ComplianceReport:
self.reports[report.report_id] = report
existing = self.reports.get(report.id)
if existing is not None:
raise ValueError(f"immutable report snapshot already exists for id={report.id}")
self.reports[report.id] = report
return report
def get_report(self, report_id: str) -> Optional[ComplianceReport]:
return self.reports.get(report_id)
def save_violation(self, violation: ComplianceViolation) -> ComplianceViolation:
self.violations[violation.violation_id] = violation
self.violations[violation.id] = violation
return violation
def get_violations_by_check_run(self, check_run_id: str) -> List[ComplianceViolation]:
return [v for v in self.violations.values() if v.check_run_id == check_run_id]
def get_violations_by_run(self, run_id: str) -> List[ComplianceViolation]:
return [v for v in self.violations.values() if v.run_id == run_id]
def get_manifests_by_candidate(self, candidate_id: str) -> List[DistributionManifest]:
return [m for m in self.manifests.values() if m.candidate_id == candidate_id]
def clear_history(self) -> None:
self.check_runs.clear()
self.reports.clear()