test json field

tests/test_json_field_fastapi.py

@@ -0,0 +1,122 @@
+import uuid
+from typing import List
+
+import databases
+import pydantic
+import pytest
+import sqlalchemy
+from fastapi import FastAPI
+from starlette.testclient import TestClient
+
+import ormar
+from tests.settings import DATABASE_URL
+
+app = FastAPI()
+
+database = databases.Database(DATABASE_URL, force_rollback=True)
+metadata = sqlalchemy.MetaData()
+app.state.database = database
+
+
+@app.on_event("startup")
+async def startup() -> None:
+    database_ = app.state.database
+    if not database_.is_connected:
+        await database_.connect()
+
+
+@app.on_event("shutdown")
+async def shutdown() -> None:
+    database_ = app.state.database
+    if database_.is_connected:
+        await database_.disconnect()
+
+
+class BaseMeta(ormar.ModelMeta):
+    metadata = metadata
+    database = database
+
+
+class Thing(ormar.Model):
+    class Meta(BaseMeta):
+        tablename = "things"
+
+    id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
+    name: str = ormar.Text(default="")
+    js: pydantic.Json = ormar.JSON()
+
+
+@app.get("/things", response_model=List[Thing])
+async def read_things():
+    return await Thing.objects.order_by("name").all()
+
+
+@app.get("/things_with_sample", response_model=List[Thing])
+async def read_things():
+    await Thing(name="b", js=["asdf", "asdf", "bobby", "nigel"]).save()
+    await Thing(name="a", js="[\"lemon\", \"raspberry\", \"lime\", \"pumice\"]").save()
+    return await Thing.objects.order_by("name").all()
+
+
+@app.get("/things_with_sample_after_init", response_model=Thing)
+async def read_things():
+    thing1 = Thing()
+    thing1.name = "d"
+    thing1.js = ["js", "set", "after", "constructor"]
+    await thing1.save()
+    return thing1
+
+
+@app.post("/things", response_model=Thing)
+async def create_things(thing: Thing):
+    thing = await thing.save()
+    return thing
+
+
+@pytest.fixture(autouse=True, scope="module")
+def create_test_database():
+    engine = sqlalchemy.create_engine(DATABASE_URL)
+    metadata.create_all(engine)
+    yield
+    metadata.drop_all(engine)
+
+
+def test_read_main():
+    client = TestClient(app)
+    with client as client:
+        response = client.get("/things_with_sample")
+        assert response.status_code == 200
+
+        # check if raw response not double encoded
+        assert '["lemon","raspberry","lime","pumice"]' in response.text
+
+        # parse json and check that we get lists not strings
+        resp = response.json()
+        assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
+        assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
+
+        # create a new one
+        response = client.post("/things", json={"js": ["test", "test2"], "name": "c"})
+        assert response.json().get("js") == ["test", "test2"]
+
+        # get all with new one
+        response = client.get("/things")
+        assert response.status_code == 200
+        assert '["test","test2"]' in response.text
+        resp = response.json()
+        assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
+        assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
+        assert resp[2].get("js") == ["test", "test2"]
+
+        response = client.get("/things_with_sample_after_init")
+        assert response.status_code == 200
+        resp = response.json()
+        assert resp.get("js") == ["js", "set", "after", "constructor"]
+
+        # test new with after constructor
+        response = client.get("/things")
+        resp = response.json()
+        assert resp[0].get("js") == ["lemon", "raspberry", "lime", "pumice"]
+        assert resp[1].get("js") == ["asdf", "asdf", "bobby", "nigel"]
+        assert resp[2].get("js") == ["test", "test2"]
+        assert resp[3].get("js") == ["js", "set", "after", "constructor"]

tests/test_m2m_through_fields.py

@@ -0,0 +1,65 @@
+import databases
+import pytest
+import sqlalchemy
+
+import ormar
+from tests.settings import DATABASE_URL
+
+database = databases.Database(DATABASE_URL, force_rollback=True)
+metadata = sqlalchemy.MetaData()
+
+
+class BaseMeta(ormar.ModelMeta):
+    database = database
+    metadata = metadata
+
+
+class Category(ormar.Model):
+    class Meta(BaseMeta):
+        tablename = "categories"
+
+    id: int = ormar.Integer(primary_key=True)
+    name: str = ormar.String(max_length=40)
+
+
+class PostCategory(ormar.Model):
+    class Meta(BaseMeta):
+        tablename = "posts_x_categories"
+
+    id: int = ormar.Integer(primary_key=True)
+    sort_order: int = ormar.Integer(nullable=True)
+
+
+class Post(ormar.Model):
+    class Meta(BaseMeta):
+        pass
+
+    id: int = ormar.Integer(primary_key=True)
+    title: str = ormar.String(max_length=200)
+    categories = ormar.ManyToMany(Category, through=PostCategory)
+
+
+@pytest.fixture(autouse=True, scope="module")
+async def create_test_database():
+    engine = sqlalchemy.create_engine(DATABASE_URL)
+    metadata.create_all(engine)
+    yield
+    metadata.drop_all(engine)
+
+
+@pytest.mark.asyncio
+async def test_setting_fields_on_through_model():
+    async with database:
+        # TODO: check/ modify following
+        # loading the data into model instance of though model?
+        # <- attach to other side? both sides? access by through, or add to fields?
+        # creating while adding to relation (kwargs in add?)
+        # creating in query (dividing kwargs between final and through)
+        # updating in query
+        # sorting in filter (special __through__<field_name> notation?)
+        # ordering by in order_by
+        # accessing from instance (both sides?)
+        # modifying from instance (both sides?)
+        # including/excluding in fields?
+        # allowing to change fk fields names in through model?
+        pass