164 lines
6.5 KiB
Python
164 lines
6.5 KiB
Python
"""Qualification rules and deterministic saved-source regression."""
|
|
|
|
import copy
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
from context import FIXTURES
|
|
from opportunity_intelligence.collectors.lrs import SourceError
|
|
from opportunity_intelligence.qualification.lrs import (
|
|
description_parts,
|
|
evaluate,
|
|
execute,
|
|
qualify,
|
|
reliable_expiry,
|
|
)
|
|
|
|
|
|
def record(sid='99', title='Cloud Architect'):
|
|
"""Build a minimal candidate with intentionally historical metadata."""
|
|
return {
|
|
'source_id': sid,
|
|
'url': f'https://jobs.lrs.com/job/details/{sid}',
|
|
'title': title,
|
|
'organization': ['LRS'],
|
|
'location': ['Remote'],
|
|
'employment_type': ['Full-time'],
|
|
'source_dates': {
|
|
'visible_parsed': ['2026-09-01'],
|
|
'metadata': [{'validThrough': '2017-03-18'}],
|
|
},
|
|
'quality_flags': ['historical_metadata_expiry_not_confirmed_closed'],
|
|
}
|
|
|
|
|
|
class QualificationTests(unittest.TestCase):
|
|
def test_historical_metadata_does_not_exclude(self):
|
|
r = evaluate(
|
|
record(), 'Azure architecture', ['Azure architecture'], '2026-09-17'
|
|
)
|
|
self.assertIsNone(r['exclusion_reason'])
|
|
|
|
def test_only_explicit_unambiguous_past_visible_expiry(self):
|
|
self.assertTrue(reliable_expiry(
|
|
['Application deadline: 2026-09-01'], '2026-09-17'
|
|
)[0])
|
|
self.assertFalse(reliable_expiry(
|
|
['Application deadline: 2026-09-17'], '2026-09-17'
|
|
)[0])
|
|
self.assertFalse(reliable_expiry(
|
|
['Application deadline: 2026-99-99'], '2026-09-17'
|
|
)[0])
|
|
self.assertFalse(reliable_expiry(
|
|
['Application deadline: 2026-09-01', 'Closing date: 2026-10-01'],
|
|
'2026-09-17',
|
|
)[0])
|
|
r = evaluate(record(), '', ['Applications close: 2026-09-01'], '2026-09-17')
|
|
self.assertEqual(r['exclusion_reason'], 'reliable_visible_expiration')
|
|
|
|
def test_unrelated_title_and_industrial_management(self):
|
|
examples = [
|
|
('Account Executive', 'Sell AI software'),
|
|
('Program Manager', 'Manage steam turbine control projects'),
|
|
]
|
|
for title, text in examples:
|
|
result = evaluate(record(title=title), text, [text], '2026-09-17')
|
|
self.assertEqual(result['exclusion_reason'], 'clearly_unrelated_role')
|
|
|
|
def test_hiring_ai_boilerplate_not_relevance(self):
|
|
text, parts = description_parts(
|
|
'<div class="job-details"><p>Gardener duties.</p>'
|
|
'<p>Generative artificial intelligence in support of our hiring '
|
|
'processes.</p></div>'
|
|
)
|
|
r = evaluate(record(title='Gardener'), text, parts, '2026-09-17')
|
|
self.assertEqual(r['exclusion_reason'], 'outside_target_scope')
|
|
|
|
def test_preserve_contract_conflicts_and_full_residency_evidence(self):
|
|
parts = [
|
|
'This is a contract position.',
|
|
'Candidates must reside in Wisconsin or Illinois.',
|
|
'Candidates must have authorization without sponsorship.',
|
|
'Required: 5 years Azure experience.',
|
|
'The pay is $85 - $110 per hour.',
|
|
'Colorado Pay Range: 80.00 - 100.00/per Hour',
|
|
]
|
|
r = evaluate(record(), ' '.join(parts), parts, '2026-09-17')
|
|
self.assertEqual(r['group'], 'Contract')
|
|
self.assertIn('header_full_time_description_contract', r['conflicts'])
|
|
self.assertEqual(len(r['advertised_pay']), 2)
|
|
self.assertIn(parts[1], r['restrictions'])
|
|
self.assertIn(parts[3], r['qualifications'])
|
|
|
|
def test_employment_and_direct_hire_conflicting_contract_language(self):
|
|
r = evaluate(
|
|
record(), '',
|
|
['Direct hire opportunity.', 'Pay for this contract position.'],
|
|
'2026-09-17',
|
|
)
|
|
self.assertEqual(r['group'], 'Employment')
|
|
self.assertIn('direct_hire_and_contract_prose_conflict', r['conflicts'])
|
|
|
|
def test_exact_duplicates_and_pipeline_preserved(self):
|
|
rows = [record(), record(), record('100')]
|
|
pipeline = [dict(record('100'), status='excluded', rationale='Reviewed')]
|
|
original = copy.deepcopy(pipeline)
|
|
descriptions = {
|
|
'99': ('Azure', ['Azure']),
|
|
'100': ('Different', ['Different']),
|
|
}
|
|
selected, audit, counts = qualify(
|
|
rows, pipeline, descriptions, '2026-09-17'
|
|
)
|
|
self.assertEqual(counts['exact_duplicates'], 1)
|
|
self.assertEqual(counts['pipeline_matches'], 1)
|
|
self.assertEqual(len(selected), 1)
|
|
self.assertEqual(audit[-1]['pipeline_matches'][0]['status'], 'excluded')
|
|
self.assertEqual(pipeline, original)
|
|
|
|
def test_different_pay_is_not_exact_duplicate_and_limit_is_audited(self):
|
|
rows = [record('99'), record('100')]
|
|
descriptions = {
|
|
'99': ('Pay $80', ['Cloud architecture']),
|
|
'100': ('Pay $90', ['Cloud architecture']),
|
|
}
|
|
selected, audit, counts = qualify(rows, [], descriptions, '2026-09-17', 1)
|
|
self.assertEqual(counts['exact_duplicates'], 0)
|
|
self.assertEqual(counts['deferred'], 1)
|
|
self.assertEqual(audit[1]['decision'], 'deferred_shortlist_limit')
|
|
self.assertEqual(selected[0]['status'], 'unreviewed')
|
|
with self.assertRaises(SourceError):
|
|
qualify(rows, [], descriptions, '2026-09-17', 16)
|
|
|
|
def test_missing_description_fails_visibly(self):
|
|
with self.assertRaises(SourceError):
|
|
description_parts('<h1>Missing</h1>')
|
|
|
|
def test_saved_run_reproducible_and_pipeline_unchanged(self):
|
|
run = FIXTURES / 'lrs-run'
|
|
pipeline = FIXTURES / 'opportunities.json'
|
|
before = pipeline.read_bytes()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
first, second = pathlib.Path(tmp) / 'one', pathlib.Path(tmp) / 'two'
|
|
summary = execute(run, pipeline, first)
|
|
execute(run, pipeline, second)
|
|
self.assertEqual(summary['counts']['input'], 84)
|
|
self.assertLessEqual(summary['counts']['shortlisted'], 15)
|
|
for file in first.iterdir():
|
|
self.assertEqual(
|
|
file.read_bytes(), (second / file.name).read_bytes()
|
|
)
|
|
audit = json.loads((first / 'audit.json').read_text(encoding='utf-8'))
|
|
self.assertEqual(len(audit), 84)
|
|
self.assertTrue(all(
|
|
result['exclusion_reason'] for result in audit
|
|
if result['decision'] == 'excluded'
|
|
))
|
|
self.assertEqual(pipeline.read_bytes(), before)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|