import json import time import unittest from unittest import mock import contact_enrichment class EnrichmentMappingTest(unittest.TestCase): def test_basic_contact_mapping(self): redmine = { "id": 42, "first_name": "Alice", "middle_name": "Marie", "last_name": "Nguyen", "company": "Acme Inc", "job_title": "Engineer", "email": "alice@acme.com", "phone": "+1 555-123-4567", "website": "https://acme.com", "birthday": "1990-03-15", "background": "VIP customer", "tag_list": ["customer", "vip"], "updated_at": "2026-07-01T12:00:00Z", } result = contact_enrichment.enrich_contact(redmine) e = result.enriched self.assertEqual(42, e["redmine_contact_id"]) self.assertEqual("Alice", e["first_name"]) self.assertEqual("Marie", e["middle_name"]) self.assertEqual("Nguyen", e["last_name"]) self.assertEqual("Alice Marie Nguyen", e["full_name"]) self.assertEqual("Acme Inc", e["company"]) self.assertEqual("Engineer", e["title"]) self.assertEqual([{"value": "alice@acme.com", "type": "work"}], e["emails"]) self.assertEqual([{"value": "+15551234567", "type": "work"}], e["phones"]) self.assertEqual("https://acme.com", e["website"]) self.assertEqual("1990-03-15", e["birthday"]) self.assertEqual("redmine", e["source"]) self.assertIn("customer", e["tags"]) self.assertIn("vip", e["tags"]) self.assertIn("from-redmine", e["tags"]) self.assertIn("VIP customer", e["notes"]) self.assertEqual("2026-07-01T12:00:00+00:00", e["last_redmine_update"]) def test_company_record_uses_company_as_name(self): redmine = { "id": 7, "is_company": True, "company": "Globex", "first_name": "Globex", "email": "info@globex.com", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual("Globex", result.enriched["full_name"]) self.assertEqual("Globex", result.enriched["first_name"]) def test_address_attributes_mapped(self): redmine = { "id": 3, "first_name": "Bob", "last_name": "Smith", "address_attributes": { "street": "123 Main St", "city": "Springfield", "state": "IL", "zip": "62701", "country": "USA", }, } result = contact_enrichment.enrich_contact(redmine) address = result.enriched["address"] self.assertEqual("123 Main St", address["street"]) self.assertEqual("Springfield", address["city"]) self.assertEqual("IL", address["region"]) self.assertEqual("62701", address["postal_code"]) self.assertEqual("USA", address["country"]) def test_email_extracted_from_background_when_missing(self): redmine = { "id": 9, "first_name": "Carol", "last_name": "Danvers", "background": "Reach her at carol@shield.gov for details.", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual([{"value": "carol@shield.gov", "type": "work"}], result.enriched["emails"]) def test_company_inferred_from_email_domain(self): redmine = { "id": 11, "first_name": "Diana", "last_name": "Prince", "email": "diana@themyscira.io", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual("Themyscira Inc", result.enriched["company"]) def test_custom_domain_to_company_mapping(self): config = contact_enrichment.EnrichmentConfig( domain_to_company={"themyscira.io": "Themyscira Labs"} ) redmine = { "id": 12, "first_name": "Diana", "last_name": "Prince", "email": "diana@themyscira.io", } result = contact_enrichment.enrich_contact(redmine, config) self.assertEqual("Themyscira Labs", result.enriched["company"]) def test_external_lookup_fills_company(self): config = contact_enrichment.EnrichmentConfig( external_lookup=lambda _cid: {"company": "Wayne Enterprises"} ) redmine = { "id": 13, "first_name": "Bruce", "last_name": "Wayne", "email": "bruce@wayne.com", } result = contact_enrichment.enrich_contact(redmine, config) self.assertEqual("Wayne Enterprises", result.enriched["company"]) def test_missing_contact_id_is_skipped(self): result = contact_enrichment.enrich_contact({"first_name": "Ghost"}) self.assertTrue(result.skipped) self.assertIn("Missing Redmine contact id", result.reason) class PhoneEnrichmentTest(unittest.TestCase): def test_us_phone_normalized_to_e164(self): redmine = { "id": 20, "first_name": "Eve", "last_name": "Polastri", "phone": "555.867.5309", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual([{"value": "+15558675309", "type": "work"}], result.enriched["phones"]) def test_mobile_phone_classified(self): redmine = { "id": 21, "first_name": "Frank", "last_name": "Castle", "phone": "mobile: 555-0100", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual("mobile", result.enriched["phones"][0]["type"]) def test_short_phone_is_ignored(self): redmine = { "id": 22, "first_name": "Grace", "last_name": "Hopper", "phone": "12345", } result = contact_enrichment.enrich_contact(redmine) self.assertEqual([], result.enriched["phones"]) class NotesAndTagsTest(unittest.TestCase): def test_notes_merge_background_and_custom_fields(self): redmine = { "id": 30, "first_name": "Hank", "last_name": "Pym", "background": "Shrinking tech", "custom_fields": [ {"name": "LinkedIn", "value": "https://linkedin.com/in/hankpym"}, {"name": "Source", "value": "referral"}, ], } result = contact_enrichment.enrich_contact(redmine) notes = result.enriched["notes"] self.assertIn("Shrinking tech", notes) self.assertIn("LinkedIn: https://linkedin.com/in/hankpym", notes) self.assertIn("Source: referral", notes) self.assertIn("Redmine contact: 30", notes) def test_tags_are_lowercased_and_deduplicated(self): redmine = { "id": 31, "first_name": "Irene", "last_name": "Adler", "tag_list": "Customer, VIP, customer", "project_id": 1, } result = contact_enrichment.enrich_contact(redmine) self.assertEqual(["customer", "from-redmine", "project-1", "vip"], result.enriched["tags"]) class RateLimitTest(unittest.TestCase): def test_rate_limit_wrapper_respects_max_calls(self): calls = [] def worker(x): calls.append(x) return x * 2 wrapped = contact_enrichment.with_rate_limit( worker, max_calls=2, per_seconds=1.0, max_retries=0 ) start = time.monotonic() self.assertEqual(2, wrapped(1)) self.assertEqual(4, wrapped(2)) self.assertEqual(6, wrapped(3)) elapsed = time.monotonic() - start self.assertGreaterEqual(elapsed, 0.45) self.assertEqual([1, 2, 3], calls) def test_rate_limit_wrapper_retries_transient_errors(self): source = mock.Mock(side_effect=[RuntimeError("boom"), RuntimeError("boom"), "ok"]) wrapped = contact_enrichment.with_rate_limit( source, max_calls=10, per_seconds=1.0, max_retries=3, backoff_base=0.01 ) self.assertEqual("ok", wrapped()) self.assertEqual(3, source.call_count) def test_rate_limit_error_propagates_without_retry(self): def worker(): raise contact_enrichment.RateLimitError("throttled") wrapped = contact_enrichment.with_rate_limit( worker, max_calls=10, per_seconds=1.0, max_retries=3 ) with self.assertRaises(contact_enrichment.RateLimitError): wrapped() class BatchSyncTest(unittest.TestCase): def test_build_sync_payload_skips_invalid_and_warns(self): contacts = [ {"id": 1, "first_name": "A", "last_name": "B", "email": "a@b.com"}, {"first_name": "Invalid"}, ] payload = contact_enrichment.build_sync_payload(contacts) self.assertEqual(1, payload["count"]) self.assertEqual(1, len(payload["records"])) self.assertEqual(1, len(payload["skipped"])) self.assertEqual("Missing Redmine contact id", payload["skipped"][0]["reason"]) class DateParsingTest(unittest.TestCase): def test_birthday_formats(self): for fmt, value in ( ("%Y-%m-%d", "1990-03-15"), ("%Y/%m/%d", "1990/03/15"), ("%m/%d/%Y", "03/15/1990"), ("%d/%m/%Y", "15/03/1990"), ): with self.subTest(fmt=fmt): result = contact_enrichment._format_birthday(value) self.assertEqual("1990-03-15", result) def test_invalid_birthday_returns_none(self): self.assertIsNone(contact_enrichment._format_birthday("not-a-date")) if __name__ == "__main__": unittest.main()