[python api test] #python #api #test

python_api_test.py

import unittest
import os, sys, time
import json
from subprocess import call
import requests
import csv

parentPath = os.path.abspath("../..")
if parentPath not in sys.path:
    sys.path.insert(0, parentPath)
os.chdir('../..')

headers = {'Content-Type': 'application/json'}
marathon_url = 'http://172.19.107.161:8080'

# True to omit Docker build and Mesos deploy
test_only = False

# Production 31~35
# application_url = 'http://172.28.4.31:4081'

# Develop
# application_url = 'http://172.19.112.135:8080'

# Docker for test
application_url = 'http://172.19.107.161:9081'

# Seyeong PC
# application_url = 'http://10.202.35.193:4081'


class ApplicationTestCase(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        """Define test variables and initialize app."""
        # Define variable
        if not test_only:

            # make build and release daemon_11st at port 9081
            current_working_dir = os.getcwd()
            call('make')
            call(['make', 'release'])

            # delete test-daemon-11st group if run
            try:
                # wait for shutdown
                while requests.get(application_url + '/static/healthcheck').status_code == 200:
                    requests.delete(marathon_url + '/v2/groups/test-daemon-11st?force=true')
                    print 'killing old test-daemon-11st'
                    time.sleep(3)
                print "old test-daemon-11st is down"

            except Exception as err:
                print "test-daemon-11st is not running!"

            # run test-daemon-11st
            try:
                requests.post(marathon_url + '/v2/apps', headers=headers, data=open('./marathon/test_daemon_11st.json', 'rb'))
            except Exception as err:
                print "marathon is not running!!!"
                return

            print "Start new test-daemon-11st"
            # wait for test-daemon-11st is up
            while True:
                try:
                    if requests.get(application_url + '/static/healthcheck').status_code == 200:
                        break
                except Exception as err:
                    print 'waiting for test-daemon-11st up'
                time.sleep(3)
                continue

        # print 'test-daemon-11st is up!!!'
        #call(['curl', '-X', '"DELETE"', 'http://172.19.107.161:8080/v2/groups/test-daemon-11st?force=true'])
        # call(['curl', '-X', 'POST', '-H', '"Content-Type: application/json"', 'http://172.19.107.161:8080/v2/apps -d @test_daemon_11st.json'])


    #    def tearDown(self):
    @classmethod
    def tearDownClass(cls):

        # To shut down application, uncomment below line
        # requests.delete(marathon_url + '/v2/groups/test-daemon-11st?force=true')
        print "Test is done. Application remains up"


    def test_v2_extract(self):
        test_case = [
            {
                "test_name": "Shoes category test",
                "json_request": {
                    "prod_no": "1029087474",
                    "image_url": "http://image.11st.co.kr/pd/17/0/8/7/4/7/4/1029087474_B.jpg",
                    "lctgr_no": "1001315",
                    "mctgr_no": "1001882",
                    "sctgr_no": "1011572"
                },
                "message": "OK",
                "status": "success"
            },
            {
                "test_name": "Wrong category test",
                "json_request": {
                    "prod_no": "1450178870",
                    "image_url": "http://image.11st.co.kr/pd/17/1/7/8/8/7/0/SDogU/1450178870_B.jpg",
                    "lctgr_no": "1001298",
                    "mctgr_no": "1001749",
                    "sctgr_no": "1010002"
                },
                "message": "categories are not matched",
                "status": "fail"
            },
            {
                "test_name": "Detection fail test",
                "json_request":{
                    "prod_no": "1994680603",
                    "image_url": "http://image.11st.co.kr/pd/18/6/8/0/6/0/3/ToSNj/1994680603_B.jpg",
                    "lctgr_no": "1001311",
                    "mctgr_no": "1001819",
                    "sctgr_no": "1006657",
                },
                "message": "scores and boxes are all []",
                "status": "fail"
            },
            {
                "test_name": "Bad image URL test",
                "json_request": {
                    "prod_no": "1994680603",
                    "image_url": "http://image.11st.co.kr/pd/18/6/8/0/6/0/3/ToSNj/1994680603_.jpg",
                    "lctgr_no": "1001311",
                    "mctgr_no": "1001819",
                    "sctgr_no": "1006657",
                },
                "message": "Url not found",
                "status": "fail"
            },
        ]

        for tc in test_case:
            json_request = tc["json_request"]
            r = requests.post(application_url + '/v2/extract', headers=headers,
                          data=json.dumps(json_request))
            r_json = r.json()

            # if success test case
            if tc['status'] == 'success':
                assert(r.status_code == 200)
                assert(r_json['status'] == tc['status'])
                assert(r_json['rois'][0]['total_dim'] == 2880)
                assert(len(r_json['rois']) > 0)
            else:
                assert (r_json['status'] == tc['status'])

            assert(tc['message'] in r_json['message'])

    def test_detect(self):
        test_case = [
            {
                "test_name": "correct extract test",
                "json_request": {
                    "img_path": "http://image.11st.co.kr/ak/7/4/8/2/0/4/1347748204_L300.jpg"
                },
                "message": "scores and boxes are all []",
                "status": "success"
            }
        ]

        for tc in test_case:
            json_request = tc["json_request"]
            r = requests.post(application_url + '/detect', headers=headers, data=json.dumps(json_request))
            r_json = r.json()

            if tc['status'] == 'success':
                assert(r_json['status'] == tc['status'])
                assert(r_json['roi'][0]['category'] == 'shoes')
                assert(len(r_json['roi']) > 0)
                rect = r_json['roi'][0]['rect']
                assert(len(rect) == 5)
                # check xywh
                assert(rect[0] + rect [2]) < r_json['image_size'][0]
                assert(rect[1] + rect [3]) < r_json['image_size'][1]

        test_case = [
            {
                "test_name": "correct extract test",
                "json_request": {
                    "img_path": "http://image.11st.co.kr/ak/7/4/8/2/0/4/1347748204_L300.jpg"
                },
                "message": "scores and boxes are all []",
                "status": "success"
            }
        ]

        for tc in test_case:
            json_request = tc["json_request"]
            r = requests.post(application_url + '/detect', headers=headers, data=json.dumps(json_request))
            r_json = r.json()

            if tc['status'] == 'success':
                assert(r_json['status'] == tc['status'])
                assert(r_json['roi'][0]['category'] == 'shoes')
                assert(len(r_json['roi']) > 0)
                rect = r_json['roi'][0]['rect']
                assert(len(rect) == 5)
                # check xywh
                assert(rect[0] + rect [2]) < r_json['image_size'][0]
                assert(rect[1] + rect [3]) < r_json['image_size'][1]


    def test_featExt(self):
        test_case = [
            {
                "test_name": "correct featExt test",
                "json_request": {
                    "img_path": "http://image.11st.co.kr/pd/18/7/2/9/6/3/7/EDJZP/1955729637_B.jpg",
                    "prod_no" : "1365651116", "lctgr_no": "1001311", "mctgr_no": "1001820", "sctgr_no": "1006579"
                },
                "message": "OK",
                "status": "success"
            }
        ]

        for tc in test_case:
            json_request = tc["json_request"]
            r = requests.post(application_url + '/featExt', headers=headers, data=json.dumps(json_request))
            r_json = r.json()

            if tc['status'] == 'success':
                assert(r_json['status'] == tc['status'])
                assert(r_json['total_dim'] == 2880)


    def test_detectObject(self):
        test_case = [
            {
                "test_name": "correct detectObject test",
                "json_request": { "img_path": "http://image.11st.co.kr/pd/17/6/3/0/8/8/7/8630887_B_1.gif" },
                "message": "OK",
                "status": "success"
            }
        ]

        for tc in test_case:
            json_request = tc["json_request"]
            r = requests.post(application_url + '/detectObject', headers=headers, data=json.dumps(json_request))
            r_json = r.json()

            if tc['status'] == 'success':
                assert(r_json['status'] == tc['status'])
                assert(r_json['roi'][0]['category'] == 'jacket_coat')
                assert(len(r_json['roi'][0]['rect']) == 4)
                assert(r_json['category_list'] == 'jacket_coat')


    def test_detectSearch(self):
        test_case = [
            # {
            #     "test_name": "correct detectSearch test 1",
            #     "json_request": {
            #         "img_url": "http://image.11st.co.kr/pd/18/5/6/3/9/4/8/FCBbX/1708563948_B.jpg",
            #         "category": "tshirts_shirts_blouse_hoody",
            #         "rect": [
            #             245.34951782226562,
            #             187.08090209960938,
            #             412.8267517089844,
            #             552.0321044921875,
            #             0.9519845247268677
            #         ]
            #     },
            #     "message": "OK",
            #     "status": "success"
            # },
            # {
            #     "test_name": "correct detectSearch test 2",
            #     "json_request": {
            #         "image_url": "http://image.11st.co.kr/pd/17/9/8/5/2/0/8/QFjJH/1001985208_B.jpg",
            #         "category": "pants",
            #         "rect": [
            #             163.72055053710938,
            #             8.404693603515625,
            #             176.02236938476562,
            #             443.8953857421875,
            #             0.9995929598808289
            #         ]
            #     },
            #     "message": "OK",
            #     "status": "success"
            # },
            # {
            #     "test_name": "correct detectSearch test 2",
            #     "json_request": {
            #         "local_url": "/data1/upload/milab/rtimages/LMWK0DPP4O0NVPPTi20180718.jpg",
            #         "category": "knit_cardigan",
            #           "rect": [
            #             285.8273010253906,
            #             168.8643798828125,
            #             109.56976318359375,
            #             222.42608642578125,
            #             0.9559428691864014
            #           ]
            #     },
            #     "message": "OK",
            #     "status": "success"
            # },
            # {
            #     "test_name": "correct detectSearch test 2",
            #     "json_request": {
            #         "local_url": "/data1/upload/milab/rtimages/LMWK0DPP4O0NVPPTi20180718.jpg",
            #         "category": "knit_cardigan",
            #           "rect": [
            #             285.8273010253906,
            #             168.8643798828125,
            #             109.56976318359375,
            #             222.42608642578125,
            #             0.9559428691864014
            #           ]
            #     },
            #     "message": "OK",
            #     "status": "success"
            # }
        ]

        for i in range(10):
            for tc in test_case:
                json_request = tc["json_request"]
                r = requests.post(application_url + '/detectSearch', headers=headers, data=json.dumps(json_request))
                r_json = r.json()

                if tc['status'] == 'success':
                    assert (r_json['status'] == tc['status'])
                    assert (r_json['message'] == tc['message'])
            import time
            # time.sleep(0.1)

    def test_memory(self):
        # v2_extract_data = self.prepare_v2_extract_data()
        # self.run_v2_extract(v2_extract_data)
        return


    def prepare_v2_extract_data(self):
        csv_path = "src/test/v2_extract.csv"
        with open(csv_path) as csvfile:
            reader = csv.reader(csvfile, delimiter=',')
            v2_extract_data = []
            for row in reader:
                data = {}
                data['prod_no'] = row[0]
                data['image_url'] = row[1]
                data['lctgr_no'] = row[2]
                data['mctgr_no'] = row[3]
                data['sctgr_no'] = row[4]
                v2_extract_data.append(data)
                print data
        return v2_extract_data


    def run_v2_extract(self, v2_extract_data):
        while True:
            for json_request in v2_extract_data:
                r = requests.post(application_url + '/v2/extract', headers=headers, data=json.dumps(json_request))


if __name__ == '__main__':
    unittest.main()