Files
W1---Opportunity-Intelligence/tests/test_practical.py
T

144 lines
6.6 KiB
Python

"""Practical eligibility, owner uncertainty and saved-run preservation tests."""
import copy
import json
from pathlib import Path
import tempfile
import unittest
from context import FIXTURES
from test_qualification import record
from opportunity_intelligence.collectors.lrs import SourceError
from opportunity_intelligence.qualification import practical
class PracticalTests(unittest.TestCase):
def setUp(self):
self.profile = practical.load_profile(practical.PROFILE)
def assess(self, title="Cloud Architect", location="Remote", parts=None):
candidate = record(title=title)
candidate["location"] = [location]
parts = parts or ["Lead Azure cloud architecture and modernization."]
return practical.assess(
candidate, " ".join(parts), parts, self.profile, "2026-09-17"
)
def test_remote_architecture(self):
self.assertEqual(self.assess()["disposition"], "practical_match")
def test_remote_developer_is_specialist(self):
result = self.assess("Software Developer AI")
self.assertEqual(result["role_family"], "software_development")
self.assertEqual(result["disposition"], "specialist_mismatch")
def test_onsite_outside_tennessee(self):
result = self.assess(location="Minneapolis, MN", parts=[
"Enterprise architecture.", "This role requires 4 days in office."
])
self.assertEqual(result["disposition"], "location_conflict")
def test_tennessee_onsite(self):
result = self.assess(location="Nashville, TN", parts=[
"Azure cloud architecture.", "Onsite in Nashville, TN."
])
self.assertEqual(result["disposition"], "practical_match")
def test_ambiguous_work_arrangement(self):
result = self.assess(location="Nashville, TN")
self.assertEqual(result["disposition"], "insufficient_evidence")
def test_unknown_certification_is_not_failure(self):
result = self.assess(parts=[
"Azure architecture.", "Azure Solutions Architect certification required."
])
self.assertEqual(result["disposition"], "needs_owner_review")
self.assertEqual(result["verification_gaps"][0]["fact"], "certifications")
self.assertEqual(result["known_owner_conflicts"], [])
def test_specialist_with_architecture_keywords(self):
for title in ("Salesforce Architect", "LifePRO Configuration Consultant"):
with self.subTest(title=title):
result = self.assess(title)
self.assertEqual(result["role_family"], "other_specialist_roles")
self.assertEqual(result["disposition"], "specialist_mismatch")
def test_builder_description_overrides_architect_title(self):
result = self.assess("AI/ML Platform Architect", parts=[
"AI architecture. Must have actually built or trained ML models."
])
self.assertEqual(result["role_family"], "data_science_ml_engineering")
self.assertEqual(result["disposition"], "specialist_mismatch")
def test_remote_restrictions_preserved(self):
parts = ["Azure architecture.", "Candidates must reside in Wisconsin or Illinois."]
result = self.assess(parts=parts)
self.assertEqual(result["disposition"], "needs_owner_review")
self.assertIn(parts[1], result["geography"]["residency_evidence"])
self.profile["owner_facts"]["residence_state"] = "TN"
self.assertEqual(self.assess(parts=parts)["disposition"], "location_conflict")
def test_owner_can_explicitly_permit_other_onsite_states(self):
self.profile["geography"]["approved_onsite_states"] = ["MN"]
self.assertEqual(self.assess(location="Minneapolis, MN", parts=[
"Azure architecture.", "Onsite in Minneapolis, MN."
])["disposition"], "practical_match")
def test_desired_certifications_are_labeled(self):
result = self.assess(parts=[
"Azure architecture.", "Requirements:", "Cloud leadership.",
"Strong Candidates Will Have:", "TOGAF Certification"
])
self.assertEqual(result["verification_gaps"][0]["requirement_level"],
"desired_or_unspecified")
def test_reliable_expiry_and_historical_conflicts(self):
result = self.assess(parts=["Azure architecture.",
"Application deadline: 2026-09-01"])
self.assertEqual(result["disposition"], "out_of_scope")
result = self.assess()
self.assertIn("historical_metadata_expiry_not_confirmed_closed",
result["source_conflicts"])
self.assertEqual(result["hiring_status"], "unconfirmed")
def test_malformed_profile_fails_visibly(self):
profile = copy.deepcopy(self.profile)
profile["owner_facts"]["certifications"] = True
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "profile.json"
path.write_text(json.dumps(profile), encoding="utf-8")
with self.assertRaises(SourceError):
practical.load_profile(path)
def test_saved_run_reproducible_and_reviewed_pipeline_unchanged(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
pipeline = root / "opportunities.json"
original = (FIXTURES / "opportunities.json").read_bytes()
pipeline.write_bytes(original)
outputs = [root / "first", root / "second"]
for output in outputs:
summary = practical.execute(
FIXTURES / "lrs-run", pipeline, practical.PROFILE, output
)
self.assertEqual(summary["input_count"], 87)
self.assertEqual(summary["existing_pipeline_records"], 3)
self.assertTrue(summary["pipeline_unchanged"])
self.assertEqual(pipeline.read_bytes(), original)
for path in outputs[0].iterdir():
self.assertEqual(path.read_bytes(), (outputs[1] / path.name).read_bytes())
records = json.loads((outputs[0] / "practical.json").read_text())
existing = [r for r in records if r["existing_pipeline_records"]]
self.assertEqual(len(existing), 3)
self.assertTrue(all(not r["reviewable_new_candidate"] for r in existing))
self.assertTrue(all(r["status"] == "unreviewed" for r in records))
with self.assertRaises(FileExistsError):
practical.execute(
FIXTURES / "lrs-run", pipeline, practical.PROFILE, outputs[0]
)
if __name__ == "__main__":
unittest.main()