Add tests for additional Django database interactions (#2)

* add tests for basic model CRUD operations
* add migration tests
* add tests for db operations like joins, aggregations, and more

Author: jcarbaugh

tests/test_integration.py

@@ -1,6 +1,7 @@
 import pytest
 from django.contrib.auth.models import User
 from django.contrib.contenttypes.models import ContentType
+from django.db.models.sql import DeleteQuery, InsertQuery, UpdateQuery
 
 from django_fqq.schema import clear_schema, set_schema
 
@@ -42,6 +43,56 @@ def test_schema_switching_between_queries():
     assert '"beta"."django_content_type"' in sql_beta
 
 
+def test_delete_query_has_schema():
+    set_schema("tenant1")
+    dq = DeleteQuery(ContentType)
+    sql = str(dq)
+    assert '"tenant1"."django_content_type"' in sql
+
+
+def test_delete_query_no_schema():
+    dq = DeleteQuery(ContentType)
+    sql = str(dq)
+    assert '"django_content_type"' in sql
+    assert '"public"."django_content_type"' not in sql
+
+
+def test_update_query_has_schema():
+    set_schema("tenant1")
+    uq = UpdateQuery(ContentType)
+    uq.add_update_values({"app_label": "test"})
+    sql = str(uq)
+    assert '"tenant1"."django_content_type"' in sql
+
+
+def test_update_query_no_schema():
+    uq = UpdateQuery(ContentType)
+    uq.add_update_values({"app_label": "test"})
+    sql = str(uq)
+    assert '"django_content_type"' in sql
+    assert '"public"."django_content_type"' not in sql
+
+
+def test_insert_query_has_schema():
+    set_schema("tenant1")
+    iq = InsertQuery(ContentType)
+    obj = ContentType(app_label="test", model="testmodel")
+    iq.insert_values(ContentType._meta.local_concrete_fields, [obj])
+    compiler = iq.get_compiler("default")
+    sql, _ = compiler.as_sql()[0]
+    assert '"tenant1"."django_content_type"' in sql
+
+
+def test_insert_query_no_schema():
+    iq = InsertQuery(ContentType)
+    obj = ContentType(app_label="test", model="testmodel")
+    iq.insert_values(ContentType._meta.local_concrete_fields, [obj])
+    compiler = iq.get_compiler("default")
+    sql, _ = compiler.as_sql()[0]
+    assert '"django_content_type"' in sql
+    assert '"public"."django_content_type"' not in sql
+
+
 # --- Shared app: auth ---
 
 
@@ -59,3 +110,28 @@ def test_shared_app_ignores_schema_switch():
     set_schema("beta")
     sql_beta = str(User.objects.all().query)
     assert '"public"."auth_user"' in sql_beta
+
+
+def test_shared_delete_uses_public_schema():
+    set_schema("tenant1")
+    dq = DeleteQuery(User)
+    sql = str(dq)
+    assert '"public"."auth_user"' in sql
+
+
+def test_shared_update_uses_public_schema():
+    set_schema("tenant1")
+    uq = UpdateQuery(User)
+    uq.add_update_values({"username": "test"})
+    sql = str(uq)
+    assert '"public"."auth_user"' in sql
+
+
+def test_shared_insert_uses_public_schema():
+    set_schema("tenant1")
+    iq = InsertQuery(User)
+    obj = User(username="test")
+    iq.insert_values(User._meta.local_concrete_fields, [obj])
+    compiler = iq.get_compiler("default")
+    sql, _ = compiler.as_sql()[0]
+    assert '"public"."auth_user"' in sql

tests/test_migrations.py

@@ -0,0 +1,165 @@
+from unittest.mock import patch
+
+import pytest
+from django.db import connection, models
+from django.db.models import Index
+
+from django_fqq.backend.operations import DatabaseOperations
+from django_fqq.schema import clear_schema, set_schema
+
+
+class TenantModel(models.Model):
+    name = models.CharField(max_length=100)
+
+    class Meta:
+        app_label = "testapp"
+        db_table = "testapp_tenantmodel"
+
+
+class TenantModelWithAge(models.Model):
+    name = models.CharField(max_length=100)
+    age = models.IntegerField(default=0)
+
+    class Meta:
+        app_label = "testapp"
+        db_table = "testapp_tenantmodel"
+
+
+@pytest.fixture(autouse=True)
+def schema_cleanup():
+    yield
+    clear_schema()
+
+
+@pytest.fixture(autouse=True)
+def register_table():
+    DatabaseOperations._table_names.add("testapp_tenantmodel")
+    yield
+    DatabaseOperations._table_names.discard("testapp_tenantmodel")
+    DatabaseOperations._table_names.discard("testapp_newname")
+
+
+def _noop_compose_sql(sql, params, connection):
+    """Substitute params into SQL without needing a real DB cursor."""
+    if params:
+        return sql % tuple(repr(p) for p in params)
+    return sql
+
+
+def _collect_sql(fn):
+    """Run schema editor operation and return collected SQL statements."""
+    with patch(
+        "django.db.backends.postgresql.operations.mogrify", side_effect=_noop_compose_sql
+    ):
+        editor = connection.schema_editor(collect_sql=True, atomic=False)
+        with editor:
+            fn(editor)
+        return editor.collected_sql
+
+
+# --- CREATE TABLE ---
+
+
+def test_create_table_has_schema():
+    set_schema("tenant1")
+    sqls = _collect_sql(lambda e: e.create_model(TenantModel))
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+def test_create_table_no_schema():
+    sqls = _collect_sql(lambda e: e.create_model(TenantModel))
+    assert any('"testapp_tenantmodel"' in s for s in sqls)
+    assert not any('"public"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- DROP TABLE ---
+
+
+def test_drop_table_has_schema():
+    set_schema("tenant1")
+    sqls = _collect_sql(lambda e: e.delete_model(TenantModel))
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+def test_drop_table_no_schema():
+    sqls = _collect_sql(lambda e: e.delete_model(TenantModel))
+    assert any('"testapp_tenantmodel"' in s for s in sqls)
+    assert not any('"public"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- ADD COLUMN ---
+
+
+def test_add_column_has_schema():
+    set_schema("tenant1")
+    field = TenantModelWithAge._meta.get_field("age")
+    sqls = _collect_sql(lambda e: e.add_field(TenantModelWithAge, field))
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+def test_add_column_no_schema():
+    field = TenantModelWithAge._meta.get_field("age")
+    sqls = _collect_sql(lambda e: e.add_field(TenantModelWithAge, field))
+    assert any('"testapp_tenantmodel"' in s for s in sqls)
+    assert not any('"public"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- DROP COLUMN ---
+
+
+def test_drop_column_has_schema():
+    set_schema("tenant1")
+    field = TenantModelWithAge._meta.get_field("age")
+    sqls = _collect_sql(lambda e: e.remove_field(TenantModelWithAge, field))
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- ADD INDEX ---
+
+
+def test_add_index_has_schema():
+    set_schema("tenant1")
+    idx = Index(fields=["name"], name="testapp_tenantmodel_name_idx")
+    sqls = _collect_sql(lambda e: e.add_index(TenantModel, idx))
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+def test_add_index_no_schema():
+    idx = Index(fields=["name"], name="testapp_tenantmodel_name_idx")
+    sqls = _collect_sql(lambda e: e.add_index(TenantModel, idx))
+    assert any('"testapp_tenantmodel"' in s for s in sqls)
+    assert not any('"public"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- RENAME TABLE ---
+
+
+def test_rename_table_has_schema():
+    set_schema("tenant1")
+    DatabaseOperations._table_names.add("testapp_newname")
+    sqls = _collect_sql(
+        lambda e: e.alter_db_table(TenantModel, "testapp_tenantmodel", "testapp_newname")
+    )
+    assert any('"tenant1"."testapp_tenantmodel"' in s for s in sqls)
+
+
+def test_rename_table_no_schema():
+    DatabaseOperations._table_names.add("testapp_newname")
+    sqls = _collect_sql(
+        lambda e: e.alter_db_table(TenantModel, "testapp_tenantmodel", "testapp_newname")
+    )
+    assert any('"testapp_tenantmodel"' in s for s in sqls)
+    assert not any('"public"."testapp_tenantmodel"' in s for s in sqls)
+
+
+# --- Schema switching across operations ---
+
+
+def test_schema_switch_between_operations():
+    set_schema("alpha")
+    sqls_alpha = _collect_sql(lambda e: e.create_model(TenantModel))
+    assert any('"alpha"."testapp_tenantmodel"' in s for s in sqls_alpha)
+
+    set_schema("beta")
+    sqls_beta = _collect_sql(lambda e: e.create_model(TenantModel))
+    assert any('"beta"."testapp_tenantmodel"' in s for s in sqls_beta)