141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Host-only contract model for the blocked Phase-0.9B observer design.
|
|
|
|
This module performs no filesystem or network I/O. It is not PS5 observer
|
|
source and must never be added to a target build.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import hashlib
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MockObject:
|
|
path_known: bool = True
|
|
path_conflict: bool = False
|
|
symlink: bool = False
|
|
object_id_before: str = "dev:1/ino:1"
|
|
object_id_after: str = "dev:1/ino:1"
|
|
size_before: int = 0
|
|
size_after: int = 0
|
|
chunks: tuple[bytes, ...] = ()
|
|
read_error: bool = False
|
|
expected_sha256: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class ObserverMock:
|
|
output_limit: int = 64
|
|
error_limit: int = 8
|
|
emitted: list[dict[str, object]] = field(default_factory=list)
|
|
errors: int = 0
|
|
retry_count: int = 0
|
|
persistent_write_count: int = 0
|
|
service_or_process_mutation_count: int = 0
|
|
listener_count: int = 0
|
|
lifecycle_call_count: int = 0
|
|
installer_call_count: int = 0
|
|
graphics_or_retroarch_call_count: int = 0
|
|
exit_reached: bool = False
|
|
|
|
def emit(self, category: str, confidence: str, raw_error: str | None) -> bool:
|
|
if len(self.emitted) >= self.output_limit:
|
|
self.errors += 1
|
|
return False
|
|
if raw_error is not None:
|
|
self.errors += 1
|
|
if self.errors > self.error_limit:
|
|
return False
|
|
self.emitted.append(
|
|
{
|
|
"category": category,
|
|
"confidence": confidence,
|
|
"raw_error": raw_error,
|
|
"persistent_mutation_performed": False,
|
|
"retry_performed": False,
|
|
}
|
|
)
|
|
return True
|
|
|
|
def finish(self) -> None:
|
|
self.exit_reached = True
|
|
|
|
|
|
def evaluate_firmware(source_one: str | None, source_two: str | None) -> str:
|
|
if source_one is None or source_two is None:
|
|
return "UNPROVEN"
|
|
if source_one != source_two:
|
|
return "CONFLICT"
|
|
return "OBSERVED"
|
|
|
|
|
|
def evaluate_object(obj: MockObject) -> tuple[str, str | None]:
|
|
if not obj.path_known:
|
|
return "UNPROVEN", "UNKNOWN_PATH"
|
|
if obj.path_conflict:
|
|
return "CONFLICT", "PATH_CONFLICT"
|
|
if obj.symlink:
|
|
return "UNPROVEN", "PATH_SYMLINK_SAFETY_UNPROVEN"
|
|
if obj.object_id_before != obj.object_id_after:
|
|
return "ERROR", "OBJECT_ID_CHANGED"
|
|
if obj.size_before != obj.size_after:
|
|
return "ERROR", "SIZE_CHANGED"
|
|
if obj.read_error:
|
|
return "ERROR", "READ_ERROR"
|
|
|
|
data = b"".join(obj.chunks)
|
|
if len(data) != obj.size_before:
|
|
return "ERROR", "SHORT_READ"
|
|
digest = hashlib.sha256(data).hexdigest()
|
|
if obj.expected_sha256 is not None and digest != obj.expected_sha256:
|
|
return "CONFLICT", "HASH_MISMATCH"
|
|
return "OBSERVED", None
|
|
|
|
|
|
def evaluate_live_backup(live: MockObject, backup: MockObject | None) -> str:
|
|
if backup is None:
|
|
return "BACKUP_MISSING"
|
|
if live.object_id_before == backup.object_id_before:
|
|
return "SAME_OBJECT"
|
|
return "SEPARATE_OBJECTS"
|
|
|
|
|
|
def unsupported_query(supported: bool) -> str:
|
|
return "OBSERVED" if supported else "UNSUPPORTED_OR_UNPROVEN"
|
|
|
|
|
|
def evaluate_autoload(source_present: bool, parse_ok: bool) -> str:
|
|
if not source_present:
|
|
return "UNPROVEN"
|
|
return "OBSERVED" if parse_ok else "ERROR"
|
|
|
|
|
|
def run_terminal_scenario(
|
|
*,
|
|
output_channel_ok: bool = True,
|
|
deadline_reached: bool = False,
|
|
record_count: int = 1,
|
|
error_count: int = 0,
|
|
) -> ObserverMock:
|
|
observer = ObserverMock()
|
|
if not output_channel_ok:
|
|
observer.emit("output", "ERROR", "OUTPUT_CHANNEL_FAILED")
|
|
observer.finish()
|
|
return observer
|
|
if deadline_reached:
|
|
observer.emit("deadline", "ERROR", "DEADLINE_REACHED")
|
|
observer.finish()
|
|
return observer
|
|
|
|
for index in range(record_count):
|
|
if not observer.emit(f"record-{index}", "OBSERVED", None):
|
|
break
|
|
for index in range(error_count):
|
|
if not observer.emit(f"error-{index}", "ERROR", "MOCK_ERROR"):
|
|
break
|
|
observer.finish()
|
|
return observer
|