200 lines
9.3 KiB
Python
200 lines
9.3 KiB
Python
"""Evidence calibration, unknown personal facts and conservative matching."""
|
|
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
from context import FIXTURES
|
|
from test_qualification import record
|
|
from opportunity_intelligence.qualification import evidence, practical
|
|
|
|
|
|
class CalibrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.profile = practical.load_profile(practical.PROFILE)
|
|
|
|
def assess(self, parts, title="Cloud Architect", location="Remote"):
|
|
candidate = record(title=title)
|
|
candidate["location"] = [location]
|
|
return practical.assess(
|
|
candidate, " ".join(parts), parts, self.profile, "2026-09-17"
|
|
)
|
|
|
|
def test_established_architecture_years(self):
|
|
result = self.assess(["Must have 15+ years architecture experience."])
|
|
self.assertFalse(result["verification_gaps"])
|
|
self.assertEqual(result["qualification_evidence"][0]["status"], "supported")
|
|
self.assertEqual(result["qualification_evidence"][0]["owner_evidence"][0]["id"],
|
|
"architecture")
|
|
|
|
def test_established_leadership_years(self):
|
|
result = self.assess(["Must have 8+ years architecture leadership."])
|
|
self.assertFalse(result["verification_gaps"])
|
|
self.assertTrue(any(
|
|
item["status"] == "supported" for item in result["qualification_evidence"]
|
|
))
|
|
|
|
def test_azure_cloud_and_strategy_have_provenance(self):
|
|
result = self.assess([
|
|
"Required Qualifications", "Azure cloud architecture.",
|
|
"Enterprise architecture strategy and roadmaps; technology modernization."
|
|
])
|
|
ids = {
|
|
owner["id"] for decision in result["qualification_evidence"]
|
|
for owner in decision["owner_evidence"]
|
|
}
|
|
self.assertTrue({"cloud", "enterprise_strategy", "modernization"} <= ids)
|
|
self.assertFalse(result["verification_gaps"])
|
|
|
|
def test_work_authorization_stays_unknown(self):
|
|
result = self.assess([
|
|
"Azure architecture.", "Must have permanent authorization to work in the USA for any employer."
|
|
])
|
|
self.assertEqual(result["disposition"], "needs_owner_review")
|
|
self.assertEqual(result["verification_gaps"][0]["owner_value"], "unknown")
|
|
|
|
def test_certification_stays_unknown_even_with_cloud_experience(self):
|
|
result = self.assess([
|
|
"Azure architecture.", "Preferred Qualifications",
|
|
"Microsoft Certified: Azure Solutions Architect Expert"
|
|
])
|
|
self.assertEqual(result["verification_gaps"][0]["fact"], "certifications")
|
|
self.assertEqual(result["verification_gaps"][0]["requirement_level"],
|
|
"desired_or_unspecified")
|
|
self.assertEqual(result["known_owner_conflicts"], [])
|
|
|
|
def test_ml_model_training_is_not_inferred_from_ai(self):
|
|
result = self.assess([
|
|
"AI architecture. Must have actually built or trained ML models.",
|
|
"Required Qualifications", "PyTorch and TensorFlow required."
|
|
], title="AI/ML Platform Architect")
|
|
self.assertEqual(result["disposition"], "specialist_mismatch")
|
|
skills = {gap.get("skill") for gap in result["verification_gaps"]}
|
|
self.assertTrue({"ML model building", "Python ML frameworks"} <= skills)
|
|
self.assertFalse(result["known_owner_conflicts"])
|
|
|
|
def test_compound_tenure_does_not_clear_specialist_requirement(self):
|
|
result = self.assess([
|
|
"Must have 15+ years architecture experience and PyTorch model training."
|
|
])
|
|
self.assertIn("role_specific_experience",
|
|
{gap["fact"] for gap in result["verification_gaps"]})
|
|
|
|
def test_insufficient_tenure_evidence_is_unknown_not_mismatch(self):
|
|
result = self.assess(["Must have 25+ years architecture experience."])
|
|
self.assertTrue(result["verification_gaps"])
|
|
self.assertFalse(result["known_owner_conflicts"])
|
|
|
|
def test_cisco_specialization_is_unknown(self):
|
|
result = self.assess([
|
|
"Qualifications", "5+ years of experience supporting Cisco Meraki networking solutions.",
|
|
"Azure architecture."
|
|
], title="Senior Network Engineer")
|
|
self.assertEqual(result["disposition"], "specialist_mismatch")
|
|
self.assertIn("Cisco Meraki",
|
|
{gap.get("skill") for gap in result["verification_gaps"]})
|
|
|
|
def test_relocation_not_inferred(self):
|
|
result = self.assess(
|
|
["Azure architecture.", "This role is onsite in Chicago, IL."],
|
|
location="Chicago, IL",
|
|
)
|
|
self.assertEqual(result["disposition"], "location_conflict")
|
|
self.assertEqual(self.profile["owner_facts"]["willing_to_relocate"], "unknown")
|
|
|
|
def test_explicit_exclusion_is_distinct_from_unknown(self):
|
|
self.profile["career_evidence"].append({
|
|
"id": "future_owner_preference", "status": "excluded",
|
|
"statement": "Owner excludes this work.", "source": "test-only owner instruction",
|
|
"level": "preference", "aliases": ["cloud architecture"],
|
|
})
|
|
result = self.assess(["Cloud architecture required."])
|
|
self.assertEqual(result["disposition"], "out_of_scope")
|
|
self.assertTrue(result["known_owner_conflicts"])
|
|
|
|
def test_credential_boilerplate_does_not_ask_personal_question(self):
|
|
result = self.assess([
|
|
"Azure architecture.", "Assist with application certification and deployment.",
|
|
"Compensation decisions are based on certifications and market considerations."
|
|
])
|
|
self.assertFalse(result["verification_gaps"])
|
|
|
|
def test_aws_familiarity_never_satisfies_specialist_tenure(self):
|
|
result = self.assess(["Must have 10+ years AWS CDK experience."])
|
|
self.assertTrue(result["verification_gaps"])
|
|
self.assertFalse(any(
|
|
item["status"] == "supported" for item in result["qualification_evidence"]
|
|
))
|
|
|
|
def test_invalid_evidence_and_legacy_profile(self):
|
|
legacy = copy.deepcopy(self.profile)
|
|
legacy["schema_version"] = 1
|
|
del legacy["career_evidence"]
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "profile.json"
|
|
path.write_text(json.dumps(legacy))
|
|
self.assertEqual(practical.load_profile(path)["schema_version"], 1)
|
|
broken = copy.deepcopy(self.profile)
|
|
broken["career_evidence"][0]["status"] = "assumed"
|
|
path.write_text(json.dumps(broken))
|
|
with self.assertRaises(ValueError):
|
|
practical.load_profile(path)
|
|
|
|
|
|
def test_comparison_requires_same_candidates(self):
|
|
with self.assertRaises(ValueError):
|
|
evidence.compare_results([{"source_id": "1"}], [])
|
|
|
|
def test_salesforce_credentials_and_depth_remain_unknown(self):
|
|
result = self.assess([
|
|
"Azure architecture.", "Required Qualifications",
|
|
"Required - Minimum of 2 Salesforce Architect Certifications",
|
|
], title="Salesforce Architect")
|
|
self.assertEqual(result["disposition"], "specialist_mismatch")
|
|
self.assertEqual({gap["fact"] for gap in result["verification_gaps"]},
|
|
{"certifications", "specialist_skills"})
|
|
|
|
def test_saved_replay_supports_tenure_and_preserves_pipeline(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
pipeline = root / "pipeline.json"
|
|
original = (FIXTURES / "opportunities.json").read_bytes()
|
|
pipeline.write_bytes(original)
|
|
first, second = root / "first", root / "second"
|
|
for output in (first, second):
|
|
practical.execute(FIXTURES / "lrs-run", pipeline,
|
|
practical.PROFILE, output)
|
|
self.assertEqual(original, pipeline.read_bytes())
|
|
for file in first.iterdir():
|
|
self.assertEqual(file.read_bytes(), (second / file.name).read_bytes())
|
|
compared = root / "compared"
|
|
practical.execute(FIXTURES / "lrs-run", pipeline, practical.PROFILE,
|
|
compared, compare_to=first)
|
|
comparison = json.loads(
|
|
(compared / "comparison.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(comparison["disposition_changed_ids"], [])
|
|
self.assertEqual(comparison["gap_or_disposition_changes"], [])
|
|
rows = json.loads((first / "practical.json").read_text(encoding="utf-8"))
|
|
target = next(row for row in rows if row["source_id"] == "46277")
|
|
self.assertEqual(target["disposition"], "needs_owner_review")
|
|
self.assertEqual(sum(
|
|
item["status"] == "supported" for item in target["qualification_evidence"]
|
|
), 2)
|
|
self.assertNotIn("role_specific_experience",
|
|
{gap["fact"] for gap in target["verification_gaps"]})
|
|
questions = json.loads(
|
|
(first / "owner-verification.json").read_text(encoding="utf-8")
|
|
)
|
|
keys = [question["fact"] for question in questions]
|
|
self.assertEqual(len(keys), len(set(keys)))
|
|
self.assertNotIn("security_clearance", keys)
|
|
self.assertTrue(all(question["affected_count"] > 0 for question in questions))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|