From 0fb61bb18e9ead90dfcc7d1a721c24a7f5100445 Mon Sep 17 00:00:00 2001 From: Ckk3 Date: Sun, 24 Nov 2024 20:05:18 +0000 Subject: [PATCH] add seconday tables tests in test_loader --- tests/conftest.py | 206 +++++++++++++++++++++++++++++++++ tests/relay/test_connection.py | 206 --------------------------------- tests/test_loader.py | 190 +++++++++++++++++++++++------- 3 files changed, 351 insertions(+), 251 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 7c600f3..b81dac6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -111,3 +111,209 @@ def async_sessionmaker(async_engine): @pytest.fixture def base(): return orm.declarative_base() + + +@pytest.fixture +def secondary_tables(base): + EmployeeDepartmentJoinTable = sqlalchemy.Table( + "employee_department_join_table", + base.metadata, + sqlalchemy.Column("employee_id", sqlalchemy.ForeignKey("employee.id"), primary_key=True), + sqlalchemy.Column("department_id", sqlalchemy.ForeignKey( + "department.id"), primary_key=True), + ) + + class Employee(base): + __tablename__ = "employee" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + role = sqlalchemy.Column(sqlalchemy.String, nullable=True) + department = orm.relationship( + "Department", + secondary="employee_department_join_table", + back_populates="employees", + ) + + class Department(base): + __tablename__ = "department" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=True) + employees = orm.relationship( + "Employee", + secondary="employee_department_join_table", + back_populates="department", + ) + + return Employee, Department + + +@pytest.fixture +def secondary_tables_with_another_foreign_key(base): + EmployeeDepartmentJoinTable = sqlalchemy.Table( + "employee_department_join_table", + base.metadata, + sqlalchemy.Column("employee_name", sqlalchemy.ForeignKey("employee.name"), primary_key=True), + sqlalchemy.Column("department_id", sqlalchemy.ForeignKey( + "department.id"), primary_key=True), + ) + + class Employee(base): + __tablename__ = "employee" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False, primary_key=True) + role = sqlalchemy.Column(sqlalchemy.String, nullable=True) + department = orm.relationship( + "Department", + secondary="employee_department_join_table", + back_populates="employees", + ) + + class Department(base): + __tablename__ = "department" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=True) + employees = orm.relationship( + "Employee", + secondary="employee_department_join_table", + back_populates="department", + ) + + return Employee, Department + + +@pytest.fixture +def secondary_tables_with_more_secondary_tables(base): + EmployeeDepartmentJoinTable = sqlalchemy.Table( + "employee_department_join_table", + base.metadata, + sqlalchemy.Column("employee_id", sqlalchemy.ForeignKey("employee.id"), primary_key=True), + sqlalchemy.Column("department_id", sqlalchemy.ForeignKey("department.id"), primary_key=True), + ) + + EmployeeBuildingJoinTable = sqlalchemy.Table( + "employee_building_join_table", + base.metadata, + sqlalchemy.Column("employee_id", sqlalchemy.ForeignKey("employee.id"), primary_key=True), + sqlalchemy.Column("building_id", sqlalchemy.ForeignKey("building.id"), primary_key=True), + ) + + class Employee(base): + __tablename__ = "employee" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + role = sqlalchemy.Column(sqlalchemy.String, nullable=True) + department = orm.relationship( + "Department", + secondary="employee_department_join_table", + back_populates="employees", + ) + building = orm.relationship( + "Building", + secondary="employee_building_join_table", + back_populates="employees", + ) + + class Department(base): + __tablename__ = "department" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + employees = orm.relationship( + "Employee", + secondary="employee_department_join_table", + back_populates="department", + ) + + class Building(base): + __tablename__ = "building" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + employees = orm.relationship( + "Employee", + secondary="employee_building_join_table", + back_populates="building", + ) + + return Employee, Department, Building + + +@pytest.fixture +def secondary_tables_with_use_list_false(base): + EmployeeDepartmentJoinTable = sqlalchemy.Table( + "employee_department_join_table", + base.metadata, + sqlalchemy.Column("employee_id", sqlalchemy.ForeignKey("employee.id"), primary_key=True), + sqlalchemy.Column("department_id", sqlalchemy.ForeignKey( + "department.id"), primary_key=True), + ) + + class Employee(base): + __tablename__ = "employee" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + role = sqlalchemy.Column(sqlalchemy.String, nullable=True) + department = orm.relationship( + "Department", + secondary="employee_department_join_table", + back_populates="employees", + ) + + class Department(base): + __tablename__ = "department" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + employees = orm.relationship( + "Employee", + secondary="employee_department_join_table", + back_populates="department", + uselist=False + ) + + return Employee, Department + + +@pytest.fixture +def secondary_tables_with_normal_relationship(base): + EmployeeDepartmentJoinTable = sqlalchemy.Table( + "employee_department_join_table", + base.metadata, + sqlalchemy.Column("employee_id", sqlalchemy.ForeignKey("employee.id"), primary_key=True), + sqlalchemy.Column("department_id", sqlalchemy.ForeignKey( + "department.id"), primary_key=True), + ) + + class Employee(base): + __tablename__ = "employee" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + role = sqlalchemy.Column(sqlalchemy.String, nullable=True) + department = orm.relationship( + "Department", + secondary="employee_department_join_table", + back_populates="employees", + ) + building_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey("building.id")) + building = orm.relationship( + "Building", + back_populates="employees", + ) + + class Department(base): + __tablename__ = "department" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + employees = orm.relationship( + "Employee", + secondary="employee_department_join_table", + back_populates="department", + ) + + class Building(base): + __tablename__ = "building" + id = sqlalchemy.Column(sqlalchemy.Integer, autoincrement=True, primary_key=True) + name = sqlalchemy.Column(sqlalchemy.String, nullable=False) + employees = orm.relationship( + "Employee", + back_populates="building", + ) + + return Employee, Department, Building diff --git a/tests/relay/test_connection.py b/tests/relay/test_connection.py index 6c18837..2f39bfc 100644 --- a/tests/relay/test_connection.py +++ b/tests/relay/test_connection.py @@ -764,40 +764,6 @@ class Query: } -@pytest.fixture -def secondary_tables(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.id"), primary_key=True), - Column("department_id", ForeignKey( - "department.id"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - role = Column(String, nullable=False) - department = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="department", - ) - - return Employee, Department - - @pytest.mark.asyncio async def test_query_with_secondary_table_with_values_list_without_list_connection( secondary_tables, @@ -1107,40 +1073,6 @@ class Query: } -@pytest.fixture -def secondary_tables_with_another_foreign_key(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_name", ForeignKey("employee.name"), primary_key=True), - Column("department_name", ForeignKey( - "department.name"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - id = Column(Integer, autoincrement=True) - name = Column(String, nullable=False, primary_key=True) - role = Column(String, nullable=False) - department = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - id = Column(Integer, autoincrement=True) - name = Column(String, nullable=False, primary_key=True) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="department", - ) - - return Employee, Department - - @pytest.mark.asyncio async def test_query_with_secondary_table_with_values_list_with_foreign_key_different_than_id( secondary_tables_with_another_foreign_key, @@ -1291,61 +1223,6 @@ async def departments(self) -> List[Department]: } -@pytest.fixture -def secondary_tables_with_more_secondary_tables(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.id"), primary_key=True), - Column("department_id", ForeignKey("department.id"), primary_key=True), - ) - - EmployeeBuildingJoinTable = Table( - "employee_building_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.id"), primary_key=True), - Column("building_id", ForeignKey("building.id"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - role = Column(String, nullable=False) - department = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - building = relationship( - "Building", - secondary="employee_building_join_table", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="department", - ) - - class Building(base): - __tablename__ = "building" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_building_join_table", - back_populates="building", - ) - - return Employee, Department, Building - - @pytest.mark.asyncio async def test_query_with_secondary_tables_with_more_than_2_colluns_values_list( secondary_tables_with_more_secondary_tables, @@ -1542,41 +1419,6 @@ async def departments(self) -> List[Department]: } -@pytest.fixture -def secondary_tables_with_use_list_false(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.id"), primary_key=True), - Column("department_id", ForeignKey( - "department.id"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - role = Column(String, nullable=False) - department = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="department", - uselist=False - ) - - return Employee, Department - - @pytest.mark.asyncio async def test_query_with_secondary_table( secondary_tables_with_use_list_false, @@ -1940,54 +1782,6 @@ async def employees(self) -> List[Employee]: } -@pytest.fixture -def secondary_tables_with_normal_relationship(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.id"), primary_key=True), - Column("department_id", ForeignKey( - "department.id"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - role = Column(String, nullable=False) - department = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - building_id = Column(Integer, ForeignKey("building.id")) - building = relationship( - "Building", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="department", - ) - - class Building(base): - __tablename__ = "building" - id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - back_populates="building", - ) - - return Employee, Department, Building - - @pytest.mark.asyncio async def test_query_with_secondary_table_with_values_list_and_normal_relationship( secondary_tables_with_normal_relationship, diff --git a/tests/test_loader.py b/tests/test_loader.py index 7c93359..898eda3 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -26,38 +26,6 @@ class Department(base): return Employee, Department -@pytest.fixture -def secondary_tables(base): - EmployeeDepartmentJoinTable = Table( - "employee_department_join_table", - base.metadata, - Column("employee_id", ForeignKey("employee.e_id"), primary_key=True), - Column("department_id", ForeignKey("department.d_id"), primary_key=True), - ) - - class Employee(base): - __tablename__ = "employee" - e_id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - departments = relationship( - "Department", - secondary="employee_department_join_table", - back_populates="employees", - ) - - class Department(base): - __tablename__ = "department" - d_id = Column(Integer, autoincrement=True, primary_key=True) - name = Column(String, nullable=False) - employees = relationship( - "Employee", - secondary="employee_department_join_table", - back_populates="departments", - ) - - return Employee, Department - - def test_loader_init(): loader = StrawberrySQLAlchemyLoader(bind=None) assert loader._bind is None @@ -146,9 +114,8 @@ async def test_loader_with_async_session( assert {e.name for e in employees} == {"e1"} -@pytest.mark.xfail @pytest.mark.asyncio -async def test_loader_for_secondary(engine, base, sessionmaker, secondary_tables): +async def test_loader_for_secondary_table(engine, base, sessionmaker, secondary_tables): Employee, Department = secondary_tables base.metadata.create_all(engine) @@ -157,30 +124,163 @@ async def test_loader_for_secondary(engine, base, sessionmaker, secondary_tables e2 = Employee(name="e2") d1 = Department(name="d1") d2 = Department(name="d2") - session.add(e1) - session.add(e2) - session.add(d1) - session.add(d2) + d3 = Department(name="d3") + session.add_all([e1, e2, d1, d2, d3]) session.flush() - e1.departments.append(d1) - e1.departments.append(d2) - e2.departments.append(d2) + e1.department.append(d1) + e1.department.append(d2) + e2.department.append(d2) session.commit() base_loader = StrawberrySQLAlchemyLoader(bind=session) - loader = base_loader.loader_for(Employee.departments.property) + loader = base_loader.loader_for(Employee.department.property) key = tuple( [ - getattr(e1, local.key) - for local, _ in Employee.departments.property.local_remote_pairs + getattr( + e1, str(Employee.department.property.local_remote_pairs[0][0].key)), + ] + ) + + departments = await loader.load(key) + assert {d.name for d in departments} == {"d1", "d2"} + + +@pytest.mark.asyncio +async def test_loader_for_secondary_tables_with_another_foreign_key(engine, base, sessionmaker, secondary_tables_with_another_foreign_key): + Employee, Department = secondary_tables_with_another_foreign_key + base.metadata.create_all(engine) + + with sessionmaker() as session: + e1 = Employee(name="e1") + e2 = Employee(name="e2") + d1 = Department(name="d1") + d2 = Department(name="d2") + d3 = Department(name="d3") + session.add_all([e1, e2, d1, d2, d3]) + session.flush() + + e1.department.append(d1) + e1.department.append(d2) + e2.department.append(d2) + session.commit() + + base_loader = StrawberrySQLAlchemyLoader(bind=session) + loader = base_loader.loader_for(Employee.department.property) + + key = tuple( + [ + getattr( + e1, str(Employee.department.property.local_remote_pairs[0][0].key)), + ] + ) + + departments = await loader.load(key) + assert {d.name for d in departments} == {"d1", "d2"} + + +@pytest.mark.asyncio +async def test_loader_for_secondary_tables_with_more_secondary_tables(engine, base, sessionmaker, secondary_tables_with_more_secondary_tables): + Employee, Department, Building = secondary_tables_with_more_secondary_tables + base.metadata.create_all(engine) + + with sessionmaker() as session: + e1 = Employee(name="e1") + e2 = Employee(name="e2") + d1 = Department(name="d1") + d2 = Department(name="d2") + d3 = Department(name="d3") + b1 = Building(id=2, name="Building 1") + session.add_all([e1, e2, d1, d2, d3, b1]) + session.flush() + + e1.department.append(d1) + e1.department.append(d2) + e2.department.append(d2) + b1.employees.append(e1) + b1.employees.append(e2) + session.commit() + + base_loader = StrawberrySQLAlchemyLoader(bind=session) + loader = base_loader.loader_for(Employee.department.property) + + key = tuple( + [ + getattr( + e1, str(Employee.department.property.local_remote_pairs[0][0].key)), + ] + ) + + departments = await loader.load(key) + assert {d.name for d in departments} == {"d1", "d2"} + + +@pytest.mark.asyncio +async def test_loader_for_secondary_tables_with_use_list_false(engine, base, sessionmaker, secondary_tables_with_use_list_false): + Employee, Department = secondary_tables_with_use_list_false + base.metadata.create_all(engine) + + with sessionmaker() as session: + e1 = Employee(name="e1") + e2 = Employee(name="e2") + d1 = Department(name="d1") + d2 = Department(name="d2") + session.add_all([e1, e2, d1, d2]) + session.flush() + + e1.department.append(d1) + session.commit() + + base_loader = StrawberrySQLAlchemyLoader(bind=session) + loader = base_loader.loader_for(Employee.department.property) + + key = tuple( + [ + getattr( + e1, str(Employee.department.property.local_remote_pairs[0][0].key)), ] ) + + departments = await loader.load(key) + assert {d.name for d in departments} == {"d1"} + + +@pytest.mark.asyncio +async def test_loader_for_secondary_tables_with_normal_relationship(engine, base, sessionmaker, secondary_tables_with_normal_relationship): + Employee, Department, Building = secondary_tables_with_normal_relationship + base.metadata.create_all(engine) + + with sessionmaker() as session: + e1 = Employee(name="e1") + e2 = Employee(name="e2") + d1 = Department(name="d1") + d2 = Department(name="d2") + d3 = Department(name="d3") + b1 = Building(id=2, name="Building 1") + session.add_all([e1, e2, d1, d2, d3, b1]) + session.flush() + + e1.department.append(d1) + e1.department.append(d2) + e2.department.append(d2) + b1.employees.append(e1) + b1.employees.append(e2) + session.commit() + + base_loader = StrawberrySQLAlchemyLoader(bind=session) + loader = base_loader.loader_for(Employee.department.property) + + key = tuple( + [ + getattr( + e1, str(Employee.department.property.local_remote_pairs[0][0].key)), + ] + ) + departments = await loader.load(key) assert {d.name for d in departments} == {"d1", "d2"} # TODO -# add secondary tables tests # Test exception \ No newline at end of file