added basic test for seeding api
This commit is contained in:
Joel Klimont
parent
107ba9c667
commit
4c4da544c2
1 changed files with 29 additions and 277 deletions
306
test.py
306
test.py
|
|
@ -1,285 +1,37 @@
|
|||
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)
|
||||
|
||||
|
||||
def change_double_elim_api_state(started: bool = False, goal: list = [0, 0], position: list = [0, 0, 0],
|
||||
op_position: list = [0, 0, 0], items: list = []):
|
||||
data = {
|
||||
"started": started,
|
||||
"goal": goal,
|
||||
"position": position,
|
||||
"opPosition": op_position,
|
||||
"items": items
|
||||
}
|
||||
|
||||
requests.post("http://localhost:5000/api/api/updateDoubleElimState", json=data)
|
||||
import os
|
||||
os.environ["API_URL"] = "http://192.168.0.7:5000/"
|
||||
from compLib.Api import Seeding
|
||||
|
||||
|
||||
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 == 204
|
||||
|
||||
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, status = Api.DoubleElim.get_position()
|
||||
assert type(status) is int
|
||||
assert status == 503
|
||||
assert type(response) == Api.Position
|
||||
assert 0 == response.x
|
||||
assert 0 == response.y
|
||||
assert 0 == response.degrees
|
||||
|
||||
change_double_elim_api_state(started=True, position=[50, 75, 90])
|
||||
|
||||
response, status = Api.DoubleElim.get_position()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == Api.Position
|
||||
assert 50 == response.x
|
||||
assert 75 == response.y
|
||||
assert 90 == response.degrees
|
||||
|
||||
def test_get_opponent(self):
|
||||
response, status = Api.DoubleElim.get_opponent()
|
||||
assert type(status) is int
|
||||
assert status == 503
|
||||
assert type(response) == Api.Position
|
||||
assert 0 == response.x
|
||||
assert 0 == response.y
|
||||
assert 0 == response.degrees
|
||||
|
||||
change_double_elim_api_state(started=True, op_position=[50, 75, 90])
|
||||
|
||||
response, status = Api.DoubleElim.get_opponent()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == Api.Position
|
||||
assert 50 == response.x
|
||||
assert 75 == response.y
|
||||
assert 90 == response.degrees
|
||||
|
||||
def test_get_goal(self):
|
||||
response, status = Api.DoubleElim.get_goal()
|
||||
assert type(status) is int
|
||||
assert status == 503
|
||||
assert type(response) == Api.Position
|
||||
assert 0 == response.x
|
||||
assert 0 == response.y
|
||||
assert response.degrees == -1
|
||||
|
||||
change_double_elim_api_state(started=True, goal=[50, 75])
|
||||
|
||||
response, status = Api.DoubleElim.get_goal()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == Api.Position
|
||||
assert 50 == response.x
|
||||
assert 75 == response.y
|
||||
assert response.degrees == -1
|
||||
|
||||
def test_get_items(self):
|
||||
response, status = Api.DoubleElim.get_items()
|
||||
assert type(status) is int
|
||||
assert status == 503
|
||||
assert type(response) == list
|
||||
assert response == []
|
||||
|
||||
change_double_elim_api_state(started=True)
|
||||
|
||||
response, status = Api.DoubleElim.get_items()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == list
|
||||
assert response == []
|
||||
|
||||
change_double_elim_api_state(started=True, items=[{"id": 0, "x": 50, "y": 75}])
|
||||
|
||||
response, status = Api.DoubleElim.get_items()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == list
|
||||
assert len(response) == 1
|
||||
assert response == [{"id": 0, "x": 50, "y": 75}]
|
||||
|
||||
change_double_elim_api_state(started=True, items=[{"id": 0, "x": 50, "y": 75}, {"id": 2, "x": -50, "y": -75}])
|
||||
|
||||
response, status = Api.DoubleElim.get_items()
|
||||
assert type(status) is int
|
||||
assert status == 200
|
||||
assert type(response) == list
|
||||
assert len(response) == 2
|
||||
assert response == [{"id": 0, "x": 50, "y": 75}, {"id": 2, "x": -50, "y": -75}]
|
||||
|
||||
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
|
||||
def test_all(self):
|
||||
ret, code = Seeding.get_garbage()
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_delivery()
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_material()
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.list_cargo()
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_cargo("green")
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_cargo("yellow")
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_cargo("blue")
|
||||
print(ret)
|
||||
print(code)
|
||||
ret, code = Seeding.get_cargo("red")
|
||||
print(ret)
|
||||
print(code)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
Reference in a new issue