58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
|
import os
|
||
|
|
import json
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
import psycopg
|
||
|
|
import tornado.testing
|
||
|
|
import testcontainers.postgres
|
||
|
|
|
||
|
|
import dbtests.app
|
||
|
|
|
||
|
|
|
||
|
|
class ApplicationTestCase(tornado.testing.AsyncHTTPTestCase):
|
||
|
|
def get_app(self) -> tornado.web.Application:
|
||
|
|
return dbtests.app.MyApplication()
|
||
|
|
|
||
|
|
|
||
|
|
class DatabaseTestCase(unittest.TestCase):
|
||
|
|
@classmethod
|
||
|
|
def setUpClass(cls) -> None:
|
||
|
|
cls._postgres = testcontainers.postgres.PostgresContainer(
|
||
|
|
"postgres:14", driver=None
|
||
|
|
)
|
||
|
|
cls._postgres.start()
|
||
|
|
pg_url = cls._postgres.get_connection_url()
|
||
|
|
with psycopg.connect(pg_url) as connection:
|
||
|
|
with connection.cursor() as cursor:
|
||
|
|
cursor.execute(
|
||
|
|
"CREATE TABLE counters (name TEXT PRIMARY KEY, value INTEGER DEFAULT 0)"
|
||
|
|
)
|
||
|
|
os.environ["DBTESTS_DATABASE"] = pg_url
|
||
|
|
cls.addClassCleanup(cls._postgres.stop)
|
||
|
|
|
||
|
|
|
||
|
|
class CounterTests(ApplicationTestCase, DatabaseTestCase):
|
||
|
|
def setUp(self) -> None:
|
||
|
|
super().setUp()
|
||
|
|
pg_url = self._postgres.get_connection_url()
|
||
|
|
with psycopg.connect(pg_url) as connection:
|
||
|
|
with connection.cursor() as cursor:
|
||
|
|
cursor.execute("DELETE FROM counters")
|
||
|
|
|
||
|
|
def test_counter_is_zero(self) -> None:
|
||
|
|
response = self.fetch("/counters/abc")
|
||
|
|
self.assertEqual(response.code, 200)
|
||
|
|
self.assertEqual(json.loads(response.body), 0)
|
||
|
|
|
||
|
|
def test_new_counter_returns_one(self) -> None:
|
||
|
|
response = self.fetch("/counters/abc", method="POST", body="")
|
||
|
|
self.assertEqual(response.code, 200)
|
||
|
|
self.assertEqual(json.loads(response.body), 1)
|
||
|
|
|
||
|
|
def test_get_incremented_counter(self) -> None:
|
||
|
|
self.fetch("/counters/abc", method="POST", body="")
|
||
|
|
self.fetch("/counters/abc", method="POST", body="")
|
||
|
|
self.fetch("/counters/abc", method="POST", body="")
|
||
|
|
response = self.fetch("/counters/abc")
|
||
|
|
self.assertEqual(json.loads(response.body), 3)
|