MEGCPO9566's subject reads "MEGCREQ11976 (Equestria 215)". A site name with a bare instance number was not recognised as a site reference, so the PO fell through the whole name chain and came back as "no MEGA site name found" rather than naming the site it wanted. A bracketed name with an instance number is now a site reference, and a text whose bracketed names all resolve is not flagged at all.
272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""Tests for the MEGA PO email intake parser and site matching.
|
|
|
|
Run: python3 scripts/test_po_email_intake.py
|
|
"""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from po_email_intake import ( # noqa: E402
|
|
name_variants,
|
|
parse_po_text,
|
|
resolve_site_by_name,
|
|
site_for_po,
|
|
unknown_site_name,
|
|
)
|
|
|
|
# MEGA's sites, as in scripts/data/po-email-intake/sites.json
|
|
SITES = [
|
|
{"id": "8qyqytyl62rysu1", "name": "The Hills", "code": "C26/09"},
|
|
{"id": "u6n5fye4czinw6o", "name": "Cedar Gate", "code": "C26/13"},
|
|
{"id": "v61qv4s1omqqkc6", "name": "Mega Head Office", "code": "HEAD"},
|
|
{"id": "dmqybquea190z22", "name": "Nellmapius Ext 25", "code": "C26/27"},
|
|
{"id": "1d4hqov69qzajcp", "name": "Munyaka 2", "code": "C26/14"},
|
|
{"id": "ne80o2ow1vblsnq", "name": "Cosmo City", "code": "C30/25"},
|
|
{"id": "1q3fvsiq7vv28np", "name": "Halfway Gardens", "code": "C31/25"},
|
|
{"id": "r4q8yrip5wbmxrp", "name": "Wespark", "code": "C26/10"},
|
|
{"id": "uaqxp126by39jkb", "name": "Montana", "code": "C25/41"},
|
|
{"id": "majjyyc7agr51aa", "name": "Mushroom farm", "code": "C11/23"},
|
|
{"id": "newcastle", "name": "Castle Gate", "code": "C26/08"},
|
|
]
|
|
|
|
# The real layout of a MEGA PO, trimmed but faithful to rptViewer output.
|
|
# MEGCPO9547 "Chamberlain PO" -- the site appears in the description and in the
|
|
# project code on each line, but nowhere in the email subject.
|
|
PO_9547 = """Mega Infrastructure (Pty) Ltd
|
|
Mega Park Business Park
|
|
70 Motor street
|
|
Silverton
|
|
Pretoria
|
|
0184
|
|
Telephone : (012) 803 4456
|
|
Vat Registration : 4280307796
|
|
Company Registration : 2022/615504/07
|
|
|
|
|
|
Purchase Order(s) #MEGCPO9547
|
|
Subject : MEGCREQ11948 a CO on MEGCREQ11947 Description : Tools
|
|
(Nellmapius EXT25)
|
|
|
|
|
|
|
|
|
|
To : FH CHAMBERLAINS PO Creation Date : 05/11/2026
|
|
Required By : 10/11/2026
|
|
Req No : MEGCREQ11948
|
|
Ordered By : Julius Mnisi
|
|
Code : 05CHA001 Telephone :
|
|
Address : Fax :
|
|
Email : julius@megainfra.co.za
|
|
Currency : (ZAR) South African Rand
|
|
Contact : FH CHAMBERLAINS Payment In : Current
|
|
Telephone :
|
|
Fax :
|
|
Cell Phone :
|
|
Email :
|
|
Bus No. Reg. : VAT Number :
|
|
|
|
|
|
Contact : Head office, Tel : (012) 803 4456, Fax : , Cell : (012) 803
|
|
4456
|
|
Delivery Address : Mega Park Business Park, 70 Motor street,
|
|
Silverton, Pretoria, 0184
|
|
|
|
|
|
|
|
No Material Cost Code Description Quantity Unit RateDISC% VAT% Amount
|
|
Code
|
|
|
|
1 513999 C26/27, 0001 Sail Easy Shade Kalahari 3m 1 each 694.7800 0.00 15.00 694.78
|
|
Square
|
|
2 513999 C26/27, 0001 Wheelbarrow Concrete 1 755.6500 0.00 15.00 755.65
|
|
S/Duty 0.8mm Pan 65L
|
|
3 513999 C26/27, 0001 Rope SKI 10MMx15M FGSK 1 55.5700 0.00 15.00 55.57
|
|
Rop Pluset
|
|
|
|
|
|
|
|
|
|
Sub Total Amount : 1,506.00
|
|
VAT Amount : 225.91
|
|
Total Amount : 1,731.91
|
|
|
|
Authorised Signatory : ________________________________
|
|
|
|
Subject to our terms and conditions of purchase ( Available on request )
|
|
|
|
Printed 2026-11-05 10:12 Page 1 of 1
|
|
Construction Computer Software (Pty) Ltd Ref: STD_po_DN.rpt (Std/Link) Mod: 12-March-2024
|
|
"""
|
|
|
|
|
|
class ParsePoText(unittest.TestCase):
|
|
def test_reads_the_header_fields(self):
|
|
parsed = parse_po_text(PO_9547)
|
|
self.assertEqual(parsed["po_number"], "MEGCPO9547")
|
|
self.assertEqual(parsed["supplier"], "FH CHAMBERLAINS")
|
|
self.assertEqual(parsed["description"], "Tools (Nellmapius EXT25)")
|
|
self.assertEqual(parsed["order_date"], "05/11/2026")
|
|
self.assertEqual(parsed["required_by"], "10/11/2026")
|
|
|
|
def test_reads_the_project_code_off_each_line(self):
|
|
# Kept as evidence on the filed row; it no longer decides the site.
|
|
self.assertEqual(parse_po_text(PO_9547)["cost_codes"], ["C26/27"])
|
|
|
|
def test_a_document_with_no_line_items_has_no_project_code(self):
|
|
self.assertEqual(parse_po_text("Purchase Order(s) #MEGCPO1\n")["cost_codes"], [])
|
|
|
|
|
|
class NameVariants(unittest.TestCase):
|
|
def test_drops_a_trailing_instance_number(self):
|
|
self.assertEqual(name_variants("Munyaka 2"), ["Munyaka 2", "Munyaka"])
|
|
|
|
def test_drops_a_trailing_ext_and_number(self):
|
|
self.assertEqual(
|
|
name_variants("Nellmapius Ext 25"),
|
|
["Nellmapius Ext 25", "Nellmapius"],
|
|
)
|
|
|
|
def test_a_name_without_a_qualifier_is_left_alone(self):
|
|
self.assertEqual(name_variants("Castle Gate"), ["Castle Gate"])
|
|
|
|
|
|
class ResolveSiteByName(unittest.TestCase):
|
|
"""The site is booked against the name written on the PO."""
|
|
|
|
def test_a_name_in_the_po_subject_line_resolves(self):
|
|
site = resolve_site_by_name("MEGCREQ11958 (Castle Gate)", SITES)
|
|
self.assertEqual(site["name"], "Castle Gate")
|
|
|
|
def test_a_name_in_the_po_description_resolves(self):
|
|
site = resolve_site_by_name("Tools (Nellmapius EXT25)", SITES)
|
|
self.assertEqual(site["name"], "Nellmapius Ext 25")
|
|
|
|
def test_a_loose_name_still_resolves(self):
|
|
site = resolve_site_by_name("MEGCREQ11879 Munyaka", SITES)
|
|
self.assertEqual(site["name"], "Munyaka 2")
|
|
|
|
def test_the_hills_is_still_found(self):
|
|
site = resolve_site_by_name("MEGCREQ11911 The Hills Ext12", SITES)
|
|
self.assertEqual(site["name"], "The Hills")
|
|
|
|
def test_a_supplier_name_is_not_a_site(self):
|
|
self.assertIsNone(resolve_site_by_name("Chamberlain PO", SITES))
|
|
|
|
def test_no_name_at_all_is_not_a_site(self):
|
|
self.assertIsNone(resolve_site_by_name("MEGCREQ11948 a CO on MEGCREQ11947", SITES))
|
|
|
|
def test_a_project_code_is_not_a_site_name(self):
|
|
# The code on the lines is a cost allocation, not the site.
|
|
self.assertIsNone(resolve_site_by_name("C26/27", SITES))
|
|
|
|
def test_the_name_beats_a_cost_code_naming_another_site(self):
|
|
# MEGCPO9550: a Cedar Gate delivery with four of its seven lines billed
|
|
# to Castle Gate's project code. The name decides.
|
|
self.assertEqual(
|
|
resolve_site_by_name("MEGCREQ11956 (Cedar Gate)", SITES)["name"],
|
|
"Cedar Gate",
|
|
)
|
|
|
|
def test_a_po_naming_two_sites_is_left_for_a_person(self):
|
|
self.assertIsNone(resolve_site_by_name("Munyaka 2 delivery for Castle Gate", SITES))
|
|
|
|
def test_a_different_spelling_of_the_name_still_resolves(self):
|
|
# MEGA writes "West Park" where the app has "Wespark".
|
|
site = resolve_site_by_name("MEGCREQ11960 (West Park EXT6)", SITES)
|
|
self.assertEqual(site["name"], "Wespark")
|
|
|
|
def test_a_misspelling_in_the_po_still_resolves(self):
|
|
# MEGCPO9475's own subject misspells the site: "Nellmpius EXT25".
|
|
site = resolve_site_by_name("MEGCREQ11843 (Nellmpius EXT25)", SITES)
|
|
self.assertEqual(site["name"], "Nellmapius Ext 25")
|
|
|
|
|
|
class SiteOfAPo(unittest.TestCase):
|
|
"""The order the intake uses: the PO's subject, then its description, then the email."""
|
|
|
|
def resolve(self, parsed, email_subject=""):
|
|
site, unknown = site_for_po(parsed, email_subject, SITES)
|
|
self.assertIsNone(unknown, "this PO should not name a site we do not have")
|
|
return site
|
|
|
|
def test_a_po_whose_email_subject_names_no_site_is_still_filed(self):
|
|
parsed = parse_po_text(PO_9547)
|
|
site = self.resolve(parsed, "Chamberlain PO")
|
|
self.assertIsNotNone(site, "MEGCPO9547 names no site in its email subject")
|
|
self.assertEqual(site["name"], "Nellmapius Ext 25")
|
|
|
|
def test_a_po_naming_no_site_anywhere_needs_review(self):
|
|
parsed = parse_po_text(PO_9547)
|
|
parsed["subject_line"] = "MEGCREQ11948 a CO on MEGCREQ11947"
|
|
parsed["description"] = ""
|
|
self.assertIsNone(self.resolve(parsed, "Chamberlain PO"))
|
|
|
|
def test_a_po_naming_a_site_we_do_not_have_is_not_booked_against_the_email(self):
|
|
# MEGCPO9564: the PO says Willow Park EXT98, but the email it arrived
|
|
# in is still titled after an older Cedar Gate PO. The email subject
|
|
# must not decide the site.
|
|
parsed = {
|
|
"subject_line": "MEGCREQ11972 (Willow Park EXT98)",
|
|
"description": "Water Fittings",
|
|
}
|
|
site, unknown = site_for_po(parsed, "RE: Rock Grid PO Cedar Gate", SITES)
|
|
self.assertIsNone(site, "must not be booked against the stale email subject")
|
|
self.assertEqual(unknown, "Willow Park EXT98")
|
|
|
|
def test_the_same_po_is_filed_once_the_site_exists(self):
|
|
sites = SITES + [{"id": "willow98", "name": "Willow Park Ext 98", "code": "C25/40"}]
|
|
parsed = {
|
|
"subject_line": "MEGCREQ11972 (Willow Park EXT98)",
|
|
"description": "Water Fittings",
|
|
}
|
|
site, unknown = site_for_po(parsed, "RE: Rock Grid PO Cedar Gate", sites)
|
|
self.assertIsNone(unknown)
|
|
self.assertEqual(site["name"], "Willow Park Ext 98")
|
|
|
|
|
|
class UnknownSiteName(unittest.TestCase):
|
|
"""The site a PO names in brackets that is not one of our sites."""
|
|
|
|
def test_a_site_we_do_not_have_is_named(self):
|
|
self.assertEqual(
|
|
unknown_site_name("MEGCREQ11972 (Willow Park EXT98)", SITES),
|
|
"Willow Park EXT98",
|
|
)
|
|
|
|
def test_a_site_we_have_is_not_unknown(self):
|
|
self.assertIsNone(unknown_site_name("MEGCREQ11969(Cedar Gate)", SITES))
|
|
|
|
def test_a_misspelling_of_a_site_we_have_is_not_unknown(self):
|
|
self.assertIsNone(unknown_site_name("MEGCREQ11843 (Nellmpius EXT25)", SITES))
|
|
|
|
def test_a_cost_code_in_brackets_is_not_a_site(self):
|
|
self.assertIsNone(unknown_site_name("Supply (C26/27)", SITES))
|
|
|
|
def test_a_product_spec_in_brackets_is_not_a_site(self):
|
|
self.assertIsNone(unknown_site_name("Pipe (110mm PVC)", SITES))
|
|
|
|
def test_a_number_in_brackets_is_not_a_site(self):
|
|
self.assertIsNone(unknown_site_name("Call (012) 803 4456", SITES))
|
|
|
|
def test_a_site_named_with_an_instance_number_is_named(self):
|
|
# MEGCPO9566: "MEGCREQ11976 (Equestria 215)".
|
|
self.assertEqual(
|
|
unknown_site_name("MEGCREQ11976 (Equestria 215)", SITES),
|
|
"Equestria 215",
|
|
)
|
|
|
|
def test_a_site_named_with_an_instance_number_resolves_once_it_exists(self):
|
|
sites = SITES + [{"id": "equestria", "name": "Equestria 215", "code": "C25/31"}]
|
|
self.assertIsNone(unknown_site_name("MEGCREQ11976 (Equestria 215)", sites))
|
|
|
|
def test_a_known_site_in_the_same_text_settles_it(self):
|
|
# One bracketed name we know means the text names a site we have.
|
|
self.assertIsNone(
|
|
unknown_site_name("Tools (Nellmapius EXT25) (Grade 40)", SITES)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|