Jason Thistlethwaite
365 lines
13 KiB
Python
365 lines
13 KiB
Python
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()
|