Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion legal-api/pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "legal-api"
version = "3.1.7"
version = "3.1.8"
description = ""
authors = [
{name = "thor",email = "1042854+thorwolpert@users.noreply.github.com"}
Expand Down
79 changes: 72 additions & 7 deletions legal-api/src/legal_api/reports/document_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,18 +50,34 @@
}, {
"documentType": "CORP_AFFIDAVIT",
"documentClass": "CORP"
}],
"continuationIn": [{
"documentType": "CNTA",
"documentClass": "CORP"
}],
"continuationOut": [{
"documentType": "CNTO",
"documentClass": "CORP"
}]
}
# Map to DRS document based on the default static document name set when the client does not submit a filename.
# Included for migrated minio docs where submissions with no filename were allowed.
STATIC_DOCUMENTS = {
"Unlimited Liability Corporation Information": {
"documentType": "DIRECTOR_AFFIDAVIT",
"documentClass": "CORP"
},
"Court Order": {
"documentType": "CRTO",
"documentClass": "*"
}
}
APP_JSON = "application/json"
APP_PDF = "application/pdf"
DOC_PATH = "/documents"
BEARER = "Bearer "
DS_ID_PREFIX = "DS"
DS_KEY_TOKENS = 2


class ReportTypes(BaseEnum):
Expand Down Expand Up @@ -500,10 +516,10 @@ def replace_filing_report(
return response2
return report_response


def _update_static_document(self, doc: dict, doc_list: list) -> list:
"""
Update static document urls in the document_list if a DRS match is found.
For static documents try obtaining download URL params from the file key first (DRS key).

docs: The DRS document information to match on.
doc_list: The business list of reports/documents for the filing.
Expand All @@ -516,12 +532,7 @@ def _update_static_document(self, doc: dict, doc_list: list) -> list:
for key in doc_list:
if key == "staticDocuments":
for static_doc in doc_list.get("staticDocuments"):
name: str = static_doc.get("name")
if (name and name == doc.get("name")) or (
name and STATIC_DOCUMENTS.get(name) and
doc.get("documentClass") == STATIC_DOCUMENTS[name].get("documentClass") and
doc.get("documentType") == STATIC_DOCUMENTS[name].get("documentType")
):
if self.is_static_doc_match(doc, static_doc):
static_doc["url"] = static_doc["url"] + query_params
break
elif any(
Expand All @@ -533,6 +544,60 @@ def _update_static_document(self, doc: dict, doc_list: list) -> list:
break
return doc_list

def get_url_drs_id(self, url: str) -> str:
"""
Try to extract a DRS ID from the static document url (either minio or DRS).

url: URL to use.
return: DRS ID if url ends with DRS file key.
"""
file_key: str = str(url).split("/")[-1]
if file_key:
tokens = file_key.split("-")
if (
len(tokens) == DS_KEY_TOKENS and
tokens[0] in ("COOP", "FIRM", "CORP") and
str(tokens[1]).startswith(DS_ID_PREFIX)
):
return tokens[1]
return ""


def is_static_doc_match(self, doc: dict, static_doc: dict) -> bool:
"""
Match the current DRS document with a ledger filing static document entry.
Try obtaining download URL params from the file key first (DRS key).

doc: Current DRS document to match with static document.
doc_list: The business list of reports/documents for the filing.
return: True if the DRS doc matches the static document information.
"""
if str(static_doc.get("url")).find("documentClass=") > 0:
return False
drs_id = self.get_url_drs_id(static_doc.get("url", ""))
if drs_id:
return drs_id == doc.get("identifier")
name: str = static_doc.get("name")
doc_name: str = doc.get("name", "")
if (doc_name):
if name.removesuffix(".pdf") == doc_name.removesuffix(".pdf"):
return True
if (
STATIC_DOCUMENTS.get(name) == name and
doc.get("documentType") == STATIC_DOCUMENTS[name].get("documentType")
):
return True
# Try partial matching static doc name with default name.
for key, value in STATIC_DOCUMENTS.items():
if (
name.startswith(str(key)) and
(doc_name.startswith(str(key)) or doc_name == "") and
value.get("documentType") == doc.get("documentType")
):
return True
return False


def _get_request_headers(self, account_id: str, accept_mime_type = None) -> dict:
"""
Get request headers for the DRS api call.
Expand Down
64 changes: 61 additions & 3 deletions legal-api/tests/unit/reports/test_document_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,11 +78,11 @@
"staticDocuments": [
{
"name": "Unlimited Liability Corporation Information",
"url": "https://test.com/C9900863/filings/155753/documents/static/DS0000100741"
"url": "https://test.com/C9900863/filings/155753/documents/static/CORP-DS0000100741"
},
{
"name": "20250107-Authorization1.pdf",
"url": "https://test.com/C9900863/filings/155753/documents/static/DS0000100740"
"url": "https://test.com/C9900863/filings/155753/documents/static/CORP-DS0000100740"
}
]
}
Expand Down Expand Up @@ -250,6 +250,22 @@
"url": ""
}
]
DRS_TEST_DOC = {
"consumerDocumentId": "0100000527",
"dateCreated": "2026-06-29T19:19:32+00:00",
"datePublished": "2026-06-29T00:00:00+00:00",
"documentClass": "COOP",
"documentType": "COSD",
"documentTypeDescription": "Statement of Dissolution",
"entityIdentifier": "CP1044808",
"eventIdentifier": 233801,
"identifier": "DS0000101951",
"name": "",
"url": ""
}
STATIC_URL1 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/798da472-4f2d-4299-a3ee-84388d89f9ce.pdf"
STATIC_URL2 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/CORP-DS0000101951"
STATIC_URL3 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/CORP-DS0000101952"

# testdata pattern is ({description}, {doc_data}, {drs_data}, {receipt}, {filing}, {noa}, {cert}, {static})
TEST_FILING_UPDATE_DATA = [
Expand Down Expand Up @@ -309,7 +325,25 @@
(True, "ceaseReceiver", "FILING"),
(True, "default", "FILING"),
]

# testdata pattern is ({match}, {url}, {name}, {drs_doc}, {drs_class}, {drs_type}, {drs_id}, {drs_filename})
TEST_STATIC_DOC_MATCH_DATA = [
(True, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
(True, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", ""),
(True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
(True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "FIRM", "CRTO", "DS0000101951", "Test court order"),
(True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "COOP", "CRTO", "DS0000101951", "Test court order.pdf"),
(True, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Court Order 1234"),
(True, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", ""),
(False, STATIC_URL3, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
(False, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101952", ""),
(False, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", ""),
(False, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", "Court Order.pdf"),
]
# testdata pattern is ({url}, {drs_id})
TEST_STATIC_URL_DRS_ID_DATA = [
(STATIC_URL1, ""),
(STATIC_URL2, "DS0000101951"),
]

@pytest.mark.parametrize("has_data,report_key,report_type", TEST_REPORT_META_DATA)
def test_report_meta_type(session, has_data, report_key, report_type):
Expand Down Expand Up @@ -390,6 +424,30 @@ def test_update_ledger_docs(session, desc, doc_data, drs_data, receipt, filing,
assert text_results.find(static) > 0


@pytest.mark.parametrize("match,url,name,drs_doc,drs_class,drs_type,drs_id,drs_filename", TEST_STATIC_DOC_MATCH_DATA)
def test_match_static_doc(session, match, url, name, drs_doc, drs_class, drs_type, drs_id, drs_filename):
"""Assert that DRS matching on static documents when building download URLs works as expected."""
doc = copy.deepcopy(drs_doc)
doc["documentClass"] = drs_class
doc["documentType"] = drs_type
doc["name"] = drs_filename if drs_filename else ""
if drs_id:
doc["identifier"] = drs_id
static_doc = {
"name": name,
"url": url
}
result = DocumentService().is_static_doc_match(doc, static_doc)
assert result == match


@pytest.mark.parametrize("url,drs_id", TEST_STATIC_URL_DRS_ID_DATA)
def test_url_drs_id_doc(session, url, drs_id):
"""Assert that extracting a DRS ID from a url file key when building download URLs works as expected."""
result = DocumentService().get_url_drs_id(url)
assert result == drs_id


def test_create_document(app, session, mock_bearer_token, mock_doc_service):
founding_date = datetime.now(UTC)
business = factory_business('CP1234567', founding_date=founding_date)
Expand Down