130 lines
5.0 KiB
Python
130 lines
5.0 KiB
Python
"""Offline LRS parsing, data quality and deduplication regression."""
|
|
|
|
import copy
|
|
import json
|
|
import unittest
|
|
|
|
from context import FIXTURES
|
|
from opportunity_intelligence.collectors.lrs import (
|
|
Page,
|
|
SourceError,
|
|
deduplicate,
|
|
discover,
|
|
extract,
|
|
structured,
|
|
)
|
|
|
|
ROOT = FIXTURES
|
|
URL = 'https://jobs.lrs.com/job/details/46369?shownonlrs=True'
|
|
|
|
|
|
class LRSTests(unittest.TestCase):
|
|
def test_encoded_mime_and_valid_graph(self):
|
|
value = {'@graph': [{'@type': 'JobPosting', 'title': 'Engineer'}]}
|
|
html = (
|
|
'<script type="application/ld+json">'
|
|
+ json.dumps(value) + '</script>'
|
|
)
|
|
root = Page(html).root
|
|
block = structured(root)[0]
|
|
self.assertTrue(block['valid_json'])
|
|
self.assertEqual(block['jobs'][0]['title'], 'Engineer')
|
|
|
|
def test_real_malformed_sample_and_conflicts(self):
|
|
sample = (ROOT / 'lrs-sample.html').read_text(encoding='utf-8-sig')
|
|
result = extract(sample, URL, '2026-09-17')
|
|
self.assertEqual(result['title'], 'AI Architect')
|
|
self.assertEqual(result['source_dates']['visible_parsed'], ['2026-09-03'])
|
|
block = result['structured_evidence'][0]
|
|
self.assertFalse(block['valid_json'])
|
|
metadata = block['recovered_untrusted_fields']
|
|
self.assertEqual(metadata['datePosted'], '2017-01-18')
|
|
self.assertEqual(metadata['validThrough'], '2017-03-18T00:00')
|
|
self.assertEqual(metadata['baseSalary']['value']['value'], 40)
|
|
self.assertIn(
|
|
'$85.00 to $120.00',
|
|
' '.join(result['compensation']['advertised_text']),
|
|
)
|
|
self.assertIn('posting_date_conflict', result['quality_flags'])
|
|
self.assertIn('employment_type_conflict', result['quality_flags'])
|
|
self.assertEqual(result['hiring_status'], 'unconfirmed')
|
|
self.assertEqual(result['http_availability'], 'not_checked')
|
|
|
|
def test_missing_optional_fields(self):
|
|
html = (
|
|
'<section class="job-detail">'
|
|
'<span class="card-title">Test (46369)</span></section>'
|
|
)
|
|
result = extract(html, URL, '2026-09-17')
|
|
self.assertIsNone(result['location'])
|
|
self.assertEqual(result['source_dates']['visible_parsed'], [])
|
|
self.assertIn('missing_json_ld', result['quality_flags'])
|
|
self.assertIn(
|
|
'missing_visible_compensation_advertised_text',
|
|
result['quality_flags'],
|
|
)
|
|
|
|
def test_changed_detail_and_wrong_identity_fail(self):
|
|
pages = (
|
|
'<h1>Access denied</h1>',
|
|
'<section class="job-detail">'
|
|
'<span class="card-title">Test (123)</span></section>',
|
|
)
|
|
for html in pages:
|
|
with self.assertRaises(SourceError):
|
|
extract(html, URL, '2026-09-17')
|
|
|
|
def test_independent_index_and_repeated_links(self):
|
|
html = '<section class="job-list"><input type="hidden" value="2">'
|
|
html += (
|
|
'<a href="/job/details/99?shownonlrs=True">'
|
|
'<span class="card-title">New role (99)</span></a>'
|
|
) * 2
|
|
html += '<a href="/job/details/100">Another</a></section>'
|
|
records = discover(html)
|
|
self.assertEqual([r['source_id'] for r in records], ['99', '100'])
|
|
self.assertEqual(
|
|
records[0]['url'],
|
|
'https://jobs.lrs.com/job/details/99?shownonlrs=True',
|
|
)
|
|
with self.assertRaises(SourceError):
|
|
discover(html.replace('value="2"', 'value="3"'))
|
|
with self.assertRaises(SourceError):
|
|
discover('<h1>No jobs parsed</h1>')
|
|
|
|
def test_real_listing(self):
|
|
html = (ROOT / 'lrs-listing-sample.html').read_text(encoding='utf-8-sig')
|
|
jobs = discover(html)
|
|
self.assertEqual(len(jobs), 87)
|
|
discovered_ids = {job['source_id'] for job in jobs}
|
|
self.assertGreater(len(discovered_ids - {'46369', '46371', '46373'}), 0)
|
|
|
|
def test_namespace_dedup_preserves_reviews_and_aliases(self):
|
|
pipeline = [
|
|
{
|
|
'id': 'reviewed', 'source_id': '46369', 'status': 'excluded',
|
|
'url': (
|
|
'https://jobs.codeworks-inc.com/job/details/46369?other=1'
|
|
),
|
|
},
|
|
{'source_id': '100', 'organization': 'Unrelated company'},
|
|
]
|
|
original = copy.deepcopy(pipeline)
|
|
records = [{'source_id': '46369'}, {'source_id': '100'}]
|
|
fresh, old = deduplicate(records, pipeline)
|
|
self.assertEqual(fresh, [{'source_id': '100'}])
|
|
self.assertEqual(old[0]['pipeline_records'][0]['status'], 'excluded')
|
|
self.assertEqual(pipeline, original)
|
|
with self.assertRaises(SourceError):
|
|
deduplicate([{'source_id': '100'}, {'source_id': '100'}], pipeline)
|
|
|
|
def test_broken_metadata_does_not_invent_values(self):
|
|
html = '<script type="application/ld+json">{"bad":</script>'
|
|
result = structured(Page(html).root)
|
|
self.assertFalse(result[0]['valid_json'])
|
|
self.assertEqual(result[0]['recovered_untrusted_fields'], {})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|