import argparse import json import subprocess import sys import tempfile from pathlib import Path import unittest import redmine_contacts class RedmineContactsBehaviorTest(unittest.TestCase): class FakeLoopClient: def __init__(self): self.updated = [] def helpdesk_issues_by_contact(self, contact_id, limit): return {"issues": [{"issue_id": 9001, "subject": "Sample ticket"}]} def issue_with_helpdesk(self, issue_id, message_limit=100): return {"issue": {"id": issue_id}, "helpdesk_messages": {"messages": []}} def update_contact(self, contact_id, fields): self.updated.append((contact_id, fields)) def test_filter_contacts_by_dates_and_tags(self): contacts = [ { "id": 1, "created_on": "2026-07-01T10:00:00Z", "updated_on": "2026-07-03T12:00:00Z", "tag_list": ["customer", "prep"], }, { "id": 2, "created_on": "2026-06-20T10:00:00Z", "updated_on": "2026-06-25T12:00:00Z", "tag_list": ["vendor"], }, { "id": 3, "created_on": "2026-07-02T10:00:00Z", "updated_on": "2026-07-02T12:00:00Z", "tag_list": [], }, ] filtered = redmine_contacts.filter_contacts( contacts, created_after="2026-07-01", updated_before="2026-07-03", with_tags=["customer"], without_tags=["prep"], no_tags=False, ) self.assertEqual([], [contact["id"] for contact in filtered]) filtered_no_tags = redmine_contacts.filter_contacts( contacts, created_after=None, updated_before=None, with_tags=[], without_tags=[], no_tags=True, ) self.assertEqual([3], [contact["id"] for contact in filtered_no_tags]) def test_filter_contacts_tag_matching_is_case_insensitive(self): contacts = [{"id": 1, "tag_list": ["Customer", "Needs-Enriched"]}] filtered = redmine_contacts.filter_contacts( contacts, with_tags=["customer"], without_tags=["NEEDS-ENRICHED"], ) self.assertEqual([], [contact["id"] for contact in filtered]) filtered_required_only = redmine_contacts.filter_contacts( contacts, with_tags=["CUSTOMER"], without_tags=[], ) self.assertEqual([1], [contact["id"] for contact in filtered_required_only]) def test_merge_tags_add_and_remove(self): merged = redmine_contacts.merge_tags( existing=["customer", "needs-enriched"], add_tags=["needs-nmi-sync", "customer"], remove_tags=["needs-enriched"], ) self.assertEqual(["customer", "needs-nmi-sync"], merged) def test_build_custom_field_patch_by_slug(self): existing = [ {"id": 70, "name": "Account Number", "value": "old"}, {"id": 88, "name": "X/Twitter", "value": "@old"}, ] patch = redmine_contacts.build_custom_field_values_patch( existing, {"account-number": "A-123", "x-twitter": "@new"}, ) self.assertEqual({"70": "A-123", "88": "@new"}, patch) def test_build_custom_field_patch_rejects_unknown_slug(self): with self.assertRaises(redmine_contacts.RedmineError): redmine_contacts.build_custom_field_values_patch( [{"id": 70, "name": "Account Number", "value": "old"}], {"linkedin": "https://example.com/u/abc"}, ) def test_has_contact_filters_detects_configured_filters(self): args = argparse.Namespace( created_after=None, created_before=None, updated_after="2026-07-01", updated_before=None, with_tag=[], without_tag=[], no_tags=False, ) self.assertTrue(redmine_contacts.has_contact_filters(args)) def test_has_contact_filters_false_without_filters(self): args = argparse.Namespace( created_after=None, created_before=None, updated_after=None, updated_before=None, with_tag=[], without_tag=[], no_tags=False, ) self.assertFalse(redmine_contacts.has_contact_filters(args)) def test_resolve_fetch_cache_mode_uses_filtered_when_filters_present_and_mode_auto(self): args = argparse.Namespace( cache_mode="auto", created_after=None, created_before=None, updated_after="2026-07-01", updated_before=None, with_tag=[], without_tag=[], no_tags=False, ) self.assertEqual("filtered", redmine_contacts.resolve_fetch_cache_mode(args)) def test_resolve_fetch_cache_mode_uses_full_when_no_filters_and_mode_auto(self): args = argparse.Namespace( cache_mode="auto", created_after=None, created_before=None, updated_after=None, updated_before=None, with_tag=[], without_tag=[], no_tags=False, ) self.assertEqual("full", redmine_contacts.resolve_fetch_cache_mode(args)) def test_build_update_fields_for_contact(self): contact = { "id": 42, "tag_list": ["Customer"], "custom_fields": [{"id": 70, "name": "Account Number", "value": "old"}], } args = argparse.Namespace( set=["job_title=Buyer"], add_tag=["needs-nmi-sync"], remove_tag=["Customer"], set_custom_field=["account-number=A-9"], ) fields = redmine_contacts.build_update_fields_for_contact(contact, args) self.assertEqual("Buyer", fields["job_title"]) self.assertEqual(["needs-nmi-sync"], fields["tag_list"]) self.assertEqual({"70": "A-9"}, fields["custom_field_values"]) def test_select_batch_contacts_respects_filters_and_limit(self): contacts = [ {"id": 1, "tag_list": ["customer"], "updated_on": "2026-07-01T00:00:00Z"}, {"id": 2, "tag_list": ["customer"], "updated_on": "2026-06-01T00:00:00Z"}, {"id": 3, "tag_list": ["vendor"], "updated_on": "2026-05-01T00:00:00Z"}, ] args = argparse.Namespace( created_after=None, created_before=None, updated_after=None, updated_before=None, with_tag=["customer"], without_tag=[], no_tags=False, max_contacts=1, order="as-is", ) selected = redmine_contacts.select_batch_contacts(contacts, args) self.assertEqual([1], [item["id"] for item in selected]) def test_select_batch_contacts_orders_oldest_updated_first(self): contacts = [ {"id": 10, "tag_list": ["customer"], "updated_on": "2026-07-03T00:00:00Z"}, {"id": 11, "tag_list": ["customer"], "updated_on": "2026-04-01T00:00:00Z"}, {"id": 12, "tag_list": ["customer"], "updated_on": "2026-06-01T00:00:00Z"}, ] args = argparse.Namespace( created_after=None, created_before=None, updated_after=None, updated_before=None, with_tag=["customer"], without_tag=[], no_tags=False, max_contacts=2, order="oldest-updated-first", ) selected = redmine_contacts.select_batch_contacts(contacts, args) self.assertEqual([11, 12], [item["id"] for item in selected]) def test_select_batch_contacts_oldest_updated_first_puts_missing_timestamps_last(self): contacts = [ {"id": 21, "tag_list": ["customer"], "updated_on": None}, {"id": 22, "tag_list": ["customer"], "updated_on": "2026-01-01T00:00:00Z"}, {"id": 23, "tag_list": ["customer"], "updated_on": "2026-02-01T00:00:00Z"}, ] args = argparse.Namespace( created_after=None, created_before=None, updated_after=None, updated_before=None, with_tag=["customer"], without_tag=[], no_tags=False, max_contacts=3, order="oldest-updated-first", ) selected = redmine_contacts.select_batch_contacts(contacts, args) self.assertEqual([22, 23, 21], [item["id"] for item in selected]) def test_load_loop_profile_normalizes_optional_defaults(self): with tempfile.TemporaryDirectory() as tmpdir: profile_path = Path(tmpdir) / "profile.json" profile_path.write_text( json.dumps( { "default_tag": "ai-checked", "filter_1": {"type": "builtin", "name": "fixed-ticket-limit", "limit": 3}, "filter_2": {"type": "command", "argv": ["python3", "noop.py"]}, } ), encoding="utf-8", ) profile = redmine_contacts.load_loop_profile(profile_path) self.assertEqual("ai-checked", profile["default_tag"]) self.assertEqual([], profile["remove_tags"]) self.assertEqual(False, profile["ticket_fetch"]["fetch_full_issues"]) self.assertEqual(25, profile["ticket_fetch"]["message_limit"]) def test_run_filter1_builtin_fixed_ticket_limit_returns_integer_decision(self): profile = { "filter_1": {"type": "builtin", "name": "fixed-ticket-limit", "limit": 4}, } decision = redmine_contacts.run_loop_filter1(profile, {"id": 123}) self.assertEqual(4, decision["result"]) self.assertEqual("builtin fixed-ticket-limit", decision["reason"]) def test_run_loop_filter2_command_passes_bundle_and_dry_run_env(self): with tempfile.TemporaryDirectory() as tmpdir: script_path = Path(tmpdir) / "filter2.py" script_path.write_text( "import json, os, sys\n" "payload = json.load(sys.stdin)\n" "json.dump({\n" " 'result': True,\n" " 'contact_id': payload['contact']['id'],\n" " 'dry_run': os.environ.get('REDMINE_LOOP_DRY_RUN'),\n" "}, sys.stdout)\n", encoding="utf-8", ) profile = { "filter_2": {"type": "command", "argv": [sys.executable, str(script_path)]}, } result = redmine_contacts.run_loop_filter2( profile, {"contact": {"id": 123}, "issues": []}, runtime={"dry_run": True, "base_url": "http://redmine.test", "project": "customer-service"}, ) self.assertEqual(True, result["result"]) self.assertEqual(123, result["contact_id"]) self.assertEqual("1", result["dry_run"]) def test_run_loop_for_contact_false_filter2_plans_default_tag_without_applying(self): with tempfile.TemporaryDirectory() as tmpdir: script_path = Path(tmpdir) / "filter2.py" script_path.write_text( "import json, sys\n" "json.load(sys.stdin)\n" "json.dump({'result': False, 'summary': 'noop'}, sys.stdout)\n", encoding="utf-8", ) profile = { "name": "test-loop", "default_tag": "ai-checked", "remove_tags": [], "filter_1": {"type": "builtin", "name": "fixed-ticket-limit", "limit": 1}, "ticket_fetch": {"fetch_full_issues": False, "message_limit": 25, "sanitize_text": True}, "filter_2": {"type": "command", "argv": [sys.executable, str(script_path)]}, } client = self.FakeLoopClient() outcome = redmine_contacts.run_loop_for_contact( client, profile, {"id": 123, "tag_list": ["needs-review"]}, runtime={"dry_run": True, "base_url": "http://redmine.test", "project": "customer-service"}, ) self.assertEqual("default-tag", outcome["action"]) self.assertEqual({"tag_list": ["ai-checked", "needs-review"]}, outcome["fields"]) self.assertEqual([], client.updated) def test_search_help_shows_examples_and_doc_path(self): repo = Path(__file__).resolve().parents[1] script = repo / "redmine_contacts.py" search_help = subprocess.run( [sys.executable, str(script), "search", "--help"], cwd=repo, capture_output=True, text=True, check=False, ) self.assertEqual(search_help.returncode, 0, search_help.stderr) self.assertIn("docs/redmine_contacts_search.md", search_help.stdout) self.assertIn("--with-tag promotional-spam", search_help.stdout) self.assertIn("helpdesk_ticket.contact_id", search_help.stdout) root_help = subprocess.run( [sys.executable, str(script), "--help"], cwd=repo, capture_output=True, text=True, check=False, ) self.assertEqual(root_help.returncode, 0, root_help.stderr) self.assertIn("search --help", root_help.stdout) self.assertIn("docs/redmine_contacts_search.md", root_help.stdout) if __name__ == "__main__": unittest.main()