PyCS/test/test_client.py

import io
import time
import eventlet

from test.base import BaseTestCase

from pycs.database.File import File
from pycs.database.Result import Result
from pycs.database.Label import Label
from pycs.database.Project import Project

class ClientTests(BaseTestCase):

    def _post(self, url, status_code=200, content_type=None, json=None, data=None):
        response = self.client.post(url,
            json=json,
            data=data,
            follow_redirects=True,
            content_type=content_type,
        )

        self.assertEqual(response.status_code, 200, response.get_data().decode())
        return response

    def test_project_creation(self):

        self.assertEqual(0, Project.query.count())
        self.assertEqual(0, Label.query.count())

        self._post(
            "/projects",
            json=dict(
                name="some name",
                description="some description",
                model=1,
                label=2,
                external=None,
            )
        )
        self.assertEqual(1, Project.query.count())

        project = Project.query.first()

        self.assertIsNotNone(project)
        self.assertIsNotNone(project.model)
        self.assertIsNotNone(project.label_provider)

        self.wait_for_coroutines()
        self.assertNotEqual(0, Label.query.count())

    def test_adding_file_with_result(self):

        self._post("/projects",
            json=dict(
                name="some name",
                description="some description",
                model=1,
                label=2,
                external=None,
            )
        )
        self.assertEqual(1, Project.query.count())
        project = Project.query.first()

        self.wait_for_coroutines()

        self.assertEqual(0, File.query.count())
        self._post(f"/projects/{project.id}/data",
            data=dict(file=(io.BytesIO(b"some content"), "image.jpg")),
            content_type="multipart/form-data",
        )

        self.assertEqual(1, File.query.count())
        file = File.query.first()

        self.assertEqual(0, Result.query.count())
        self._post(f"data/{file.id}/results",
            json=dict(
                type="bounding-box",
                data=dict(x0=0, x1=0, y0=0, y1=0),
                label=2,
            )
        )
        self.assertEqual(1, Result.query.count())

    def test_cascade_after_project_removal(self):

        self.assertEqual(0, File.query.count())
        self.assertEqual(0, Result.query.count())
        self.assertEqual(0, Label.query.count())
        self.assertEqual(0, Project.query.count())

        self._post("/projects",
            json=dict(
                name="some name",
                description="some description",
                model=1,
                label=2,
                external=None,
            )
        )
        project = Project.query.first()
        project_id = project.id

        self.wait_for_coroutines()
        self._post(f"/projects/{project_id}/data",
            data=dict(file=(io.BytesIO(b"some content"), "image.jpg")),
            content_type="multipart/form-data",
        )
        file = File.query.first()
        file_id = file.id

        self.wait_for_coroutines()
        self._post(f"data/{file_id}/results",
            json=dict(
                type="bounding-box",
                data=dict(x0=0, x1=0, y0=0, y1=0),
                label=2,
            )
        )


        self.assertNotEqual(0, File.query.count())
        self.assertNotEqual(0, Result.query.count())
        self.assertNotEqual(0, Label.query.count())
        self.assertNotEqual(0, Project.query.count())

        self.wait_for_coroutines()
        eventlet.sleep(3)
        self._post(f"/projects/{project_id}/remove",
            json=dict(remove=True),
        )

        self.assertEqual(0, Project.query.count())
        self.assertEqual(0, Label.query.count())
        self.assertEqual(0, File.query.count())
        self.assertEqual(0, Result.query.count())