Add AI tests

README.md

@@ -1,2 +1,18 @@
 # SBT Project
 
+## Local development
+
+- Prepare `.env` file.
+- Run for local development:
+
+```bash
+docker compose --profile local up -d
+```
+
+## Run tests
+
+```bash
+pip3 install pytest requests
+IDP_HOST=https://sbt.idp.sdsrv.ai IDP_USERNAME=<username> IDP_PASSWORD=<password> python3 -m pytest --log-cli-level=DEBUG --capture=sys tests
+```
+

tests/test_login.py

@@ -0,0 +1,12 @@
+import sys
+sys.path.append(".")
+from tests.utils import login
+import os
+
+HOST = os.environ.get("IDP_HOST", "XXXXXXXXXXXXXXXXXXXXX")
+USERNAME = os.environ.get("IDP_USERNAME", "XXXXXXXXXXXXXXXXXXXXX")
+PASSWORD = os.environ.get("IDP_PASSWORD", "XXXXXXXXXXXXXXXXXXXXX")
+
+def test_login():
+    token = login(HOST, USERNAME, PASSWORD)
+    assert token is not None

tests/test_set_01_invoice_05_imeis.py

@@ -0,0 +1,148 @@
+import sys
+sys.path.append(".")
+from tests.utils import login, make_sbt_request, get_field
+import os
+import logging
+
+HOST = os.environ.get("IDP_HOST", "XXXXXXXXXXXXXXXXXXXXX")
+USERNAME = os.environ.get("IDP_USERNAME", "XXXXXXXXXXXXXXXXXXXXX")
+PASSWORD = os.environ.get("IDP_PASSWORD", "XXXXXXXXXXXXXXXXXXXXX")
+token = login(HOST, USERNAME, PASSWORD)
+
+
+def check_invoice_data(document):
+    assert document["doc_type"] == "sbt_document"
+    assert get_field(document, "retailername") == "Arrow Communication"
+    assert get_field(document, "sold_to_party") is None
+    assert "2021-02-09" in get_field(document, "purchase_date")
+
+def check_invoice_data_2(document):
+    assert document["doc_type"] == "sbt_document"
+    assert get_field(document, "retailername") == "Starhub Shop"
+    assert get_field(document, "sold_to_party") is None
+    assert "2022-02-22" in get_field(document, "purchase_date")
+
+def test_1_invoice_1_imei():
+    invoice_files = [
+        "test_samples/sbt/invoice.jpg"
+    ]
+    imei_files = [
+        "test_samples/sbt/imei1.jpg"
+    ]
+    data = make_sbt_request(host=HOST, token=token, invoice_files=invoice_files, imei_files=imei_files)
+
+    data = data["data"]["data"]
+    assert data["total_pages"] == 2
+    assert data["ocr_num_pages"] == 2
+    assert data["status"] == 200
+
+    document = data["document"][0]
+    assert document["end_page"] == 2
+    assert document["start_page"] == 1
+    check_invoice_data(document)
+
+    assert "358975990917032" == get_field(document, "imei_number")[0]
+
+
+def test_1_invoice_3_imei():
+    invoice_files = [
+        "test_samples/sbt/invoice.jpg"
+    ]
+    imei_files = [
+        "test_samples/sbt/imei1.jpg",
+        "test_samples/sbt/imei2.jpg",
+        "test_samples/sbt/imei3.jpg"
+    ]
+    data = make_sbt_request(host=HOST, token=token, invoice_files=invoice_files, imei_files=imei_files)
+
+    data = data["data"]["data"]
+    assert data["total_pages"] == 4
+    assert data["ocr_num_pages"] == 4
+    assert data["status"] == 200
+
+    document = data["document"][0]
+    assert document["end_page"] == 4
+    assert document["start_page"] == 1
+    check_invoice_data(document)
+
+    assert 3 == len(get_field(document, "imei_number"))
+    assert "358975990917032" == get_field(document, "imei_number")[0]
+    assert "350731691693549" == get_field(document, "imei_number")[1]
+    assert "R52W70BHDWX" == get_field(document, "imei_number")[2]
+
+
+def test_1_invoice_5_imei():
+    invoice_files = [
+        "test_samples/sbt/invoice.jpg"
+    ]
+    imei_files = [
+        "test_samples/sbt/imei1.jpg",
+        "test_samples/sbt/imei2.jpg",
+        "test_samples/sbt/imei3.jpg",
+        "test_samples/sbt/imei4.jpeg",
+        "test_samples/sbt/imei5.jpg"
+    ]
+    data = make_sbt_request(host=HOST, token=token, invoice_files=invoice_files, imei_files=imei_files)
+
+    data = data["data"]["data"]
+    assert data["total_pages"] == 6
+    assert data["ocr_num_pages"] == 6
+    assert data["status"] == 200
+
+    document = data["document"][0]
+    assert document["end_page"] == 6
+    assert document["start_page"] == 1
+    check_invoice_data(document)
+
+    assert 5 == len(get_field(document, "imei_number"))
+    assert "358975990917032" == get_field(document, "imei_number")[0]
+    assert "350731691693549" == get_field(document, "imei_number")[1]
+    assert "R52W70BHDWX" == get_field(document, "imei_number")[2]
+    assert "350073345090297" == get_field(document, "imei_number")[3]
+    assert "0PBL3NHW500023N" == get_field(document, "imei_number")[4]
+
+
+def test_0_invoice_3_imei():
+    invoice_files = []
+    imei_files = [
+        "test_samples/sbt/imei1.jpg",
+        "test_samples/sbt/imei2.jpg",
+        "test_samples/sbt/imei3.jpg"
+    ]
+    data = make_sbt_request(host=HOST, token=token, invoice_files=invoice_files, imei_files=imei_files)
+
+    data = data["data"]["data"]
+    assert data["total_pages"] == 3
+    assert data["ocr_num_pages"] == 3
+    assert data["status"] == 200
+
+    document = data["document"][0]
+    assert document["end_page"] == 3
+    assert document["start_page"] == 1
+
+    assert 3 == len(get_field(document, "imei_number"))
+    assert "358975990917032" == get_field(document, "imei_number")[0]
+    assert "350731691693549" == get_field(document, "imei_number")[1]
+    assert "R52W70BHDWX" == get_field(document, "imei_number")[2]
+
+
+def test_1_invoice_pdf_1_imei():
+    invoice_files = [
+        "test_samples/sbt/20220303025923NHNE_20220222_Starhub_Order_Confirmation_by_Email.pdf"
+    ]
+    imei_files = [
+        "test_samples/sbt/imei1.jpg"
+    ]
+    data = make_sbt_request(host=HOST, token=token, invoice_files=invoice_files, imei_files=imei_files)
+
+    data = data["data"]["data"]
+    assert data["total_pages"] == 2
+    assert data["ocr_num_pages"] == 2
+    assert data["status"] == 200
+
+    document = data["document"][0]
+    assert document["end_page"] == 2
+    assert document["start_page"] == 1
+    check_invoice_data_2(document)
+
+    assert "358975990917032" == get_field(document, "imei_number")[0]

tests/utils.py

@@ -0,0 +1,90 @@
+import requests
+import traceback
+import logging
+import time
+from requests_toolbelt import  MultipartEncoderMonitor
+
+
+def login(host, username, password):
+    print(host, username, password)
+    response = requests.post(f'{host}/api/ctel/login/', json={
+        'username': username,
+        'password': password
+    })
+    try:
+        token = response.json().get('token', '')
+    except:
+        logging.error("Failed to login")
+        traceback.print_exc()
+        raise ValueError("Could not login. Please check host, username or password.")
+    if not token:
+        logging.error(response.content)
+        raise ValueError("Could not get a valid token. Please check host, username or password.")
+
+    return token
+
+
+def make_sbt_request(host, token, invoice_files=None, imei_files=None):
+    files =  []
+    if invoice_files:
+        for file in invoice_files:
+            files.append(('invoice_file', (file, open(file, "rb"), 'application/octet-stream')))
+    if imei_files:
+        for file in imei_files:
+            files.append(('imei_files', (file, open(file, "rb"), 'application/octet-stream')))
+    num_files = len(files)
+    files.append(('processType', '12'))
+    files.append(('is_test_request', 'true'))
+    start_time = time.time()
+    end_of_upload_time = 0
+
+    def upload_callback(monitor):
+        nonlocal end_of_upload_time
+        if monitor.bytes_read == monitor.len:
+            end_of_upload_time = time.time()
+    m = MultipartEncoderMonitor.from_fields(
+        fields=files,
+        callback=upload_callback
+    )
+    try:
+        response = requests.post(f'{host}/api/ctel/images/process_sync/', headers={
+            'Authorization': token,
+            'Content-Type': m.content_type
+        }, data=m, timeout=300)
+    except requests.exceptions.Timeout as e:
+        print(e)
+        traceback.print_exc()
+        raise TimeoutError("Timeout when sending request to server.")
+    except Exception as e:
+        print(e)
+        traceback.print_exc()
+        raise e
+    upload_time = end_of_upload_time - start_time
+    process_time = time.time() - start_time - upload_time
+
+
+    assert upload_time > 0
+    assert process_time > 0
+    assert num_files > 0
+    assert response.status_code == 200
+
+    parsed_data = response.json()
+    assert "id" in parsed_data
+    assert "pages" in parsed_data
+    assert "created_at" in parsed_data
+    assert "data" in parsed_data
+    assert "files" in parsed_data
+
+    return {
+        "upload_time": upload_time,
+        "process_time": process_time,
+        "data": parsed_data,
+        "num_files": num_files,
+    }
+
+
+def get_field(document, label):
+    for field in document["content"]:
+        if field["label"] == label:
+            return field["value"]
+    return None