themblem/api/tests/test_api.py

#!/usr/bin/env python3
from django.test import TestCase, Client
from django.test import Client
from products.models import *
from products.ip2region import ip_to_region
from django.contrib.auth.models import User
import subprocess
import uuid
import time
import json
import os
import atexit
from PIL import Image
import io

BASE_DIR = os.path.abspath(os.path.dirname(__file__) + "/..")

class APITest(TestCase):
    def setUp(self):
        self.admin = User(is_superuser=True, username="admin", password="testpass")
        self.admin.save()

        self.admin_token = 'admintoken123'
        ai = AdminInfo.objects.create(user=self.admin)
        AuthToken.objects.create(token=self.admin_token, admin=ai)

        self.tenant_token = 'tenanttoken0'
        t = Tenant(username="tenant0", mobile="12300000000", password="testpass")
        t.save()

        AuthToken.objects.create(token=self.tenant_token, tenant=t)

        self.tenant = t
        self.client = Client()

        self.article = Article.objects.create(title="product info", body="", tenant=self.tenant)
        self.product = Product.objects.create(name='testprod', tenant=self.tenant)
        self.batch = CodeBatch.objects.create(tenant=self.tenant, code_prefix="ASY")
        self.code = SerialCode.objects.create(code='ASY0021010915303887353C00102212',
                batch=self.batch,
                tenant=self.tenant,
                product=self.product)

    def post(self, uri, data={}, admin=False, **kwargs):
        token = self.tenant_token
        if admin:
            token = self.admin_token
        if isinstance(data, bytes):
            ct = 'application/octet-stream'
        else:
            ct = 'application/json'
        return self.client.post(uri, data=data, content_type=ct,
                HTTP_AUTHORIZATION="token " + token,
                **kwargs)

    def patch(self, uri, data, admin=False):
        token = self.tenant_token
        if admin:
            token = self.admin_token
        headers = {
            'HTTP_AUTHORIZATION': 'token ' + token,
        }
        return self.client.patch(uri, json.dumps(data), content_type="application/json", **headers)

    def get(self, uri, admin=False):
        token = self.tenant_token
        if admin:
            token = self.admin_token
        headers = {
            'HTTP_AUTHORIZATION': 'token ' + token,
        }
        return self.client.get(uri, **headers)

    def create_product(self, product_name):
        data = {
            'name': product_name,
            'description': 'test product',
        }
        r = self.post("/api/v1/product/", data)
        self.assertEqual(r.status_code, 201)

    def test_create_batch(self):
        self.create_product("product0")
        prod = Product.objects.first()
        r = self.post("/api/v1/code-batch/", {
            'product': 1,
            'qr_angle': 2,
            'code_prefix': "3",
        })

        self.assertEqual(r.status_code, 201)

    def test_healtz(self):
        r = self.client.get("/api/v1/healthz/");
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json()['status'], 'running')

    def test_ip_to_region(self):
        region = ip_to_region('4.4.4.4')
        self.assertIn("美国", region)

    def test_patch_product_article(self):
        a = self.article
        p = self.product

        r = self.get("/api/v1/article/%d/" % a.id);
        self.assertEqual(r.status_code, 200)

        r = self.patch("/api/v1/product/%d/" % p.id, {
            "article": "/api/v1/article/%d/" % a.pk,
            })
        self.assertEqual(r.status_code, 202)
        self.assertEqual(Product.objects.get(pk=p.pk).article.pk, a.pk)

    def test_batch_inactive(self):
        self.batch.is_active = False
        self.batch.save()
        self.code.is_active = True
        self.code.save()
        with open(os.path.join(BASE_DIR, "tests/data/qr1.jpg"), 'rb') as img:
            r = self.client.post("/api/v1/qr-verify/", {
                'photo': img,
                'qrcode': 'http://localhost/?code=' + self.code.code,
            })
            self.assertIn('error', r.json())
            self.assertIn('Inactive code', r.json()['error'])

    def test_code_inactive(self):
        self.batch.is_active = True
        self.batch.save()
        self.code.is_active = False
        self.code.save()
        with open(os.path.join(BASE_DIR, "tests/data/qr1.jpg"), 'rb') as img:
            r = self.client.post("/api/v1/qr-verify/", {
                'qrcode': 'http://localhost/?code=' + self.code.code,
                'photo': img,
            })
            self.assertIn('error', r.json())
            self.assertIn('Inactive code', r.json()['error'])

    def test_post_code_feature_roi(self):
        roi_file = os.path.join(BASE_DIR, 'tests/data/0074252612205-roi.jpg')
        img = Image.open(roi_file)
        width = img.width
        r = self.post("/api/v1/code-feature-roi/", admin=True,
                files={ "0074252612205.jpg": open(roi_file, 'rb') })
        self.assertEqual(r.status_code, 200)
        r = self.get("/api/v1/code-feature-roi/?code=0074252612205", admin=True)
        img = Image.open(io.BytesIO(r.content))
        self.assertEqual(img.width, width)

    def test_estor_archive(self):
        fn = '/emblem/batches/2023-7-8-22-49-10文件夹 - emblem100/import/emblem100/1259109190683.jpg'
        body = b'abcdef'
        r = self.post("/api/v1/estor-archive/", data=body,
                HTTP_X_ESTOR_PATH=fn,
                HTTP_X_ESTOR_PARAM_EMBLEM_TOKEN=self.admin_token
            )
        self.assertEqual(r.status_code, 200)
        self.assertIn('status', r.json())