compLIB/test.py

import json
import os
import time
import unittest
from threading import Thread

import requests

# compapi is a dependency for this test
try:
    from compapi import server

    START_SERVER = True
except ImportError:
    print("[!] error could not import server module from compapi, assuming server is running")
    START_SERVER = False

from multiprocessing import Process

from compLib import LogstashLogging
from compLib import Api


def change_api_state(park_id: int = 0, in_get_park: bool = False, was_in_park: bool = False,
                     simon_says_ids: list = [0, 0, 0, 0, 0],
                     in_simon_zone: bool = False):
    data = {
        "parkId": park_id,
        "inGetPark": in_get_park,
        "wasInPark": was_in_park,
        "simonSaysIDs": simon_says_ids,
        "inSimonZone": in_simon_zone
    }

    requests.post("http://localhost:5000/api/api/updateState", json=data)


class TestApiServer(unittest.TestCase):
    def test_get_park(self):
        ret, code = Api.Seeding.get_park()
        id = ret["id"]
        assert type(ret) is dict
        assert type(id) is int
        assert id == -1
        assert type(code) is int
        assert code == 403

        change_api_state(in_get_park=True)

        ret, code = Api.Seeding.get_park()
        id = ret["id"]
        assert type(ret) is dict
        assert type(id) is int
        assert 0 <= id < 4
        assert type(code) is int
        assert code == 200

    def test_pay_park(self):
        ret = Api.Seeding.pay_park()
        assert type(ret) is int
        assert ret == 402

        change_api_state(was_in_park=True)

        ret = Api.Seeding.pay_park()
        assert type(ret) is int
        assert ret == 402

        change_api_state(was_in_park=True, in_get_park=True)

        ret = Api.Seeding.pay_park()
        assert type(ret) is int
        assert ret == 201

    def test_simon_says(self):
        change_api_state(in_simon_zone=False)
        ret, code = Api.Seeding.simon_says()
        id = ret["id"]
        assert type(ret) is dict
        assert type(id) is int
        assert type(code) is int
        assert code == 403
        assert id == -1

        got = []
        while len(got) != 5:
            change_api_state(in_simon_zone=True)
            ret, code = Api.Seeding.simon_says()
            id = ret["id"]
            assert type(ret) is dict
            assert type(id) is int
            assert type(code) is int
            assert code == 200
            assert id <= 0 < 4
            got.append(ret)
            print(got)

        change_api_state(in_simon_zone=True)
        ret, code = Api.Seeding.simon_says()
        id = ret["id"]
        assert type(ret) is dict
        assert type(id) is int
        assert type(code) is int
        assert code == 200
        assert id == -1

        change_api_state(in_simon_zone=False)
        ret, code = Api.Seeding.simon_says()
        id = ret["id"]
        assert type(ret) is dict
        assert type(id) is int
        assert type(code) is int
        assert code == 403
        assert id == -1

    def test_simon_says_iterations(self):
        for i in range(0, 5):
            change_api_state(in_simon_zone=True)
            ret, status = Api.Seeding.simon_says()
            assert type(ret) is dict
            assert type(status) is int
            assert type(ret["id"]) is int
            assert 0 <= ret["id"] < 4

        # after 4 iterations the api should only return -1 from now
        for i in range(0, 10):
            assert Api.Seeding.simon_says()[0]["id"] == -1

        # after api reset this test should work again
        self.resetApi()

        for i in range(0, 4):
            change_api_state(in_simon_zone=True)
            ret, status = Api.Seeding.simon_says()
            assert type(ret) is dict
            assert type(status) is int
            assert type(ret["id"]) is int
            assert 0 <= ret["id"] < 4

        for i in range(0, 10):
            assert Api.Seeding.simon_says()[0]["id"] == -1

    def test_simon_says_non_repeat(self):
        """
        Note: this test will fail if testing against the local api as the
        configuration is to return zeros if not otherwise specified.

        Checks if simons says does not send the robot to the
        same position again.
        """

        last_pos, status = Api.Seeding.simon_says()
        last_pos = last_pos["id"]
        for i in range(0, 100):
            change_api_state(in_simon_zone=True)
            next_pos, status = Api.Seeding.simon_says()
            next_pos = next_pos["id"]
            if next_pos == -1:
                last_pos = -1  # state is reset, so reset here as well
                self.resetApi()
                continue
            assert last_pos != next_pos
            last_pos = next_pos

    def test_get_position(self):
        response = Api.DoubleElim.get_position()
        assert type(response) == Api.Position
        assert 0 <= response.x
        assert 0 <= response.y
        assert 0 <= response.degrees <= 360

    def test_get_opponent(self):
        response = Api.DoubleElim.get_opponent()
        assert type(response) == Api.Position

    def test_get_goal(self):
        response = Api.DoubleElim.get_goal()
        assert type(response) == Api.Position
        assert 0 <= response.x
        assert 0 <= response.y
        assert response.degrees == -1

    def test_get_items(self):
        response = Api.DoubleElim.get_items()
        assert type(response) == list

    def setUp(self) -> None:
        if START_SERVER:
            self.server = Process(target=server.app.run, kwargs={"host": "0.0.0.0", "port": "5000"})
            self.server.start()
            time.sleep(0.25)
        else:
            self.resetApi()

    def tearDown(self) -> None:
        if START_SERVER:
            self.server.terminate()
            self.server.join()

    def resetApi(self):
        assert requests.get(Api.API_URL + "resetState").status_code == 200


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