Coverage for app/tests/test_sync.py: 100%

1from sqlalchemy import Engine, text 

2 

3from app.sync import sync_pypi_downloads, sync_pypi_packages 

4 

5 

6def test_sync_pypi_packages(read_engine: Engine, write_engine: Engine) -> None: 

7 limit_value = 10 

8 sync_pypi_packages(read_engine, write_engine, limit_value) 

9 with read_engine.connect() as conn: 

10 result = conn.execute(text("select count(*) from pypi_packages")) 

11 num_of_rows_inserted = result.scalar() 

12 assert num_of_rows_inserted == limit_value 

13 

14 

15def test_sync_pypi_downloads(read_engine: Engine, write_engine: Engine) -> None: 

16 limit_value = 8 

17 start_date = "2024-11-23" 

18 end_date = "2024-11-23" 

19 sync_pypi_downloads(write_engine, start_date, end_date, limit_value) 

20 with read_engine.connect() as conn: 

21 result = conn.execute( 

22 text("select count(*) from pypi_package_downloads_weekly_metrics") 

23 ) 

24 num_of_rows_inserted = result.scalar() 

25 assert num_of_rows_inserted == limit_value