Function test_schema

tests/backends/test_postgres.py:32–75

(db_simple)

@pytest.mark.asyncio
async def test_schema(db_simple):
    db_config, is_asyncpg, is_psycopg = _get_db_config()
    if not is_asyncpg and not is_psycopg:
        pytest.skip("PostgreSQL only")

    if is_asyncpg:
        from asyncpg.exceptions import InvalidSchemaNameError
    else:
        from psycopg.errors import InvalidSchemaName as InvalidSchemaNameError

    if Tortoise._inited:
        await Tortoise._drop_databases()

    try:
        db_config["connections"]["models"]["credentials"]["schema"] = "mytestschema"
        await Tortoise.init(db_config, _create_db=True)

        with pytest.raises(InvalidSchemaNameError):
            await Tortoise.generate_schemas()

        conn = connections.get("models")
        await conn.execute_script("CREATE SCHEMA mytestschema;")
        await Tortoise.generate_schemas()

        tournament = await Tournament.create(name="Test")
        await connections.close_all()

        del db_config["connections"]["models"]["credentials"]["schema"]
        await Tortoise.init(db_config)

        with pytest.raises(OperationalError):
            await Tournament.filter(name="Test").first()

        conn = connections.get("models")
        _, res = await conn.execute_query(
            "SELECT id, name FROM mytestschema.tournament WHERE name='Test' LIMIT 1"
        )

        assert len(res) == 1
        assert tournament.id == res[0]["id"]
        assert tournament.name == res[0]["name"]
    finally:
        if Tortoise._inited:
            await Tortoise._drop_databases()


@pytest.mark.asyncio