Code / tests/public_research_code/test_check_projection.py
tests/public_research_code/test_check_projection.py
1
"""Focused safety checks for the curated public-research-code candidate."""
2
3
from __future__ import annotations
4
5
import hashlib
6
import json
7
import os
8
import shutil
9
import subprocess
10
import sys
11
from pathlib import Path
12
13
from public_research_code import check_projection
14
15
INVENTORY_SOURCE = Path("public_research_code/inventory.json")
16
REQUIRED_CAPABILITY_PATHS = frozenset(
17
{
18
"libs/huleedu_nlp_shared/src/huleedu_nlp_shared/feature_contracts/current.py",
19
"scripts/ml_training/essay_scoring/features/pipeline.py",
20
"scripts/ml_training/essay_scoring/features/embeddings.py",
21
"scripts/ml_training/essay_scoring/training/xgboost/trainer.py",
22
"scripts/ml_training/essay_scoring/training/prediction_mapping.py",
23
"services/cj_assessment_service/cj_core_logic/comparison_batch_orchestrator.py",
24
"services/cj_assessment_service/cj_core_logic/bt_inference.py",
25
"services/llm_provider_service/response_validator.py",
26
"public_research_code/run_synthetic_demo.py",
27
}
28
)
29
30
31
def _inventory_paths() -> list[str]:
32
source = json.loads(INVENTORY_SOURCE.read_text(encoding="utf-8"))
33
files = source["files"]
34
assert isinstance(files, list)
35
return [str(path) for path in files]
36
37
38
def _write_manifest(root: Path, paths: list[str]) -> None:
39
inventory_bytes = (
40
json.dumps({"schema_version": 1, "files": paths}, indent=2).encode("utf-8") + b"\n"
41
)
42
(root / "public_research_code/inventory.json").write_bytes(inventory_bytes)
43
manifest = {
44
"schema_version": 1,
45
"inventory_sha256": hashlib.sha256(inventory_bytes).hexdigest(),
46
"consumer_check": check_projection.EXPECTED_CHECK,
47
"limits": {"max_files": len(paths), "max_file_bytes": 100000, "max_total_bytes": 2000000},
48
}
49
(root / "public_research_code/manifest.json").write_text(
50
json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
51
)
52
53
54
def _candidate(tmp_path: Path) -> Path:
55
root = tmp_path / "candidate"
56
paths = _inventory_paths()
57
for path in paths:
58
source = Path.cwd() / path
59
target = root / path
60
target.parent.mkdir(parents=True, exist_ok=True)
61
shutil.copyfile(source, target)
62
return root
63
64
65
def _reason(root: Path) -> str:
66
reason = check_projection.validate_candidate(root)
67
assert reason is not None
68
return reason
69
70
71
def _isolated_demo_environment(root: Path) -> dict[str, str]:
72
source_root = Path.cwd().resolve()
73
third_party_paths = [
74
Path(entry).resolve()
75
for entry in sys.path
76
if entry
77
and ({"site-packages", "__pypackages__"} & set(Path(entry).parts))
78
and Path(entry).resolve() != source_root
79
]
80
allowed_paths = [
81
root,
82
root / "libs/common_core/src",
83
root / "libs/huleedu_nlp_shared/src",
84
*third_party_paths,
85
]
86
environment = {key: value for key, value in os.environ.items() if key != "PYTHONHOME"}
87
environment["PYTHONPATH"] = os.pathsep.join(str(path) for path in allowed_paths)
88
return environment
89
90
91
def test_accepts_exact_safe_candidate(tmp_path: Path) -> None:
92
assert check_projection.validate_candidate(_candidate(tmp_path)) is None
93
94
95
def test_inventory_covers_the_complete_public_execution_capabilities() -> None:
96
assert REQUIRED_CAPABILITY_PATHS <= set(_inventory_paths())
97
98
99
def test_main_accepts_projector_appended_candidate_root(tmp_path: Path) -> None:
100
assert check_projection.main(["check_projection.py", str(_candidate(tmp_path))]) == 0
101
102
103
def test_synthetic_demo_runs_from_isolated_candidate(tmp_path: Path) -> None:
104
root = _candidate(tmp_path)
105
environment = _isolated_demo_environment(root)
106
configured_paths = [
107
Path(entry).resolve() for entry in environment["PYTHONPATH"].split(os.pathsep)
108
]
109
110
assert Path.cwd().resolve() not in configured_paths
111
assert all(
112
path == root
113
or path.is_relative_to(root)
114
or {"site-packages", "__pypackages__"} & set(path.parts)
115
for path in configured_paths
116
)
117
118
result = subprocess.run(
119
[sys.executable, "-P", "public_research_code/run_synthetic_demo.py"],
120
cwd=root,
121
env=environment,
122
check=False,
123
capture_output=True,
124
text=True,
125
)
126
127
assert result.returncode == 0, result.stderr
128
payload = json.loads(result.stdout)
129
assert payload["feature_registry"] == "huleedu.scorer_features.current"
130
assert payload["mapped_predictions"] == [1.0, 1.5, 2.5, 3.0]
131
assert payload["bradley_terry_reference"] == "anchor"
132
assert check_projection.validate_candidate(root) is None
133
134
135
def test_rejects_manifest_inventory_digest_mismatch(tmp_path: Path) -> None:
136
root = _candidate(tmp_path)
137
manifest_path = root / "public_research_code/manifest.json"
138
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
139
manifest["inventory_sha256"] = "0" * 64
140
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
141
142
assert "inventory_sha256" in _reason(root)
143
144
145
def test_rejects_unexpected_and_missing_files(tmp_path: Path) -> None:
146
root = _candidate(tmp_path)
147
(root / "unexpected.txt").write_text("synthetic\n", encoding="utf-8")
148
assert "file set" in _reason(root)
149
150
(root / "unexpected.txt").unlink()
151
(root / "LICENSE").unlink()
152
assert "file set" in _reason(root)
153
154
155
def test_rejects_symlink(tmp_path: Path) -> None:
156
root = _candidate(tmp_path)
157
target = tmp_path / "outside.txt"
158
target.write_text("synthetic\n", encoding="utf-8")
159
readme = root / "public_research_code/README.md"
160
readme.unlink()
161
readme.symlink_to(target)
162
163
assert "non-regular" in _reason(root)
164
165
166
def test_rejects_invalid_utf8_and_nul_content(tmp_path: Path) -> None:
167
root = _candidate(tmp_path)
168
readme = root / "public_research_code/README.md"
169
readme.write_bytes(b"\xff")
170
assert "not UTF-8" in _reason(root)
171
172
readme.write_bytes(b"safe\x00content")
173
assert "NUL" in _reason(root)
174
175
176
def test_rejects_prohibited_path_category(tmp_path: Path) -> None:
177
root = _candidate(tmp_path)
178
paths = sorted(_inventory_paths() + ["datasets/synthetic.txt"])
179
target = root / "datasets/synthetic.txt"
180
target.parent.mkdir(parents=True)
181
target.write_text("synthetic\n", encoding="utf-8")
182
_write_manifest(root, paths)
183
184
assert "prohibited path" in _reason(root)
185
186
187
def test_allows_typed_python_models_but_rejects_model_weights(tmp_path: Path) -> None:
188
root = _candidate(tmp_path)
189
paths = sorted(_inventory_paths() + ["models/weights.bin"])
190
target = root / "models/weights.bin"
191
target.parent.mkdir(parents=True)
192
target.write_bytes(b"not a public model")
193
_write_manifest(root, paths)
194
195
assert "prohibited path" in _reason(root)
196
197
198
def test_rejects_high_confidence_credential_content(tmp_path: Path) -> None:
199
root = _candidate(tmp_path)
200
token = "ghp_" + "a" * 36
201
(root / "public_research_code/README.md").write_text(token, encoding="utf-8")
202
203
assert "credential-like" in _reason(root)
204
205
root = _candidate(tmp_path)
206
generic_secret = "token = " + '"synthetic_token_value_1234"'
207
(root / "public_research_code/README.md").write_text(generic_secret, encoding="utf-8")
208
209
assert "credential-like" in _reason(root)