Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/opam/tests/test_lister.py
| Show First 20 Lines • Show All 73 Lines • ▼ Show 20 Lines | def test_lister_opam_optional_instance(swh_scheduler): | ||||
| netloc = "opam.ocaml.org" | netloc = "opam.ocaml.org" | ||||
| instance_url = f"https://{netloc}" | instance_url = f"https://{netloc}" | ||||
| lister = OpamLister(swh_scheduler, url=instance_url,) | lister = OpamLister(swh_scheduler, url=instance_url,) | ||||
| assert lister.instance == netloc | assert lister.instance == netloc | ||||
| assert lister.opam_root == "/tmp/opam/" | assert lister.opam_root == "/tmp/opam/" | ||||
| def test_urls(swh_scheduler, mock_opam): | def test_urls(swh_scheduler, mock_opam, tmp_path): | ||||
| mock_init, mock_popen = mock_opam | mock_init, mock_popen = mock_opam | ||||
| instance_url = "https://opam.ocaml.org" | instance_url = "https://opam.ocaml.org" | ||||
| tmp_folder = mkdtemp(prefix="swh_opam_lister") | tmp_folder = mkdtemp(dir=tmp_path, prefix="swh_opam_lister") | ||||
| lister = OpamLister( | lister = OpamLister( | ||||
| swh_scheduler, url=instance_url, instance="opam", opam_root=tmp_folder, | swh_scheduler, url=instance_url, instance="opam", opam_root=tmp_folder, | ||||
| ) | ) | ||||
| assert lister.instance == "opam" | assert lister.instance == "opam" | ||||
| assert lister.opam_root == tmp_folder | assert lister.opam_root == tmp_folder | ||||
| # call the lister and get all listed origins urls | # call the lister and get all listed origins urls | ||||
| Show All 13 Lines | expected_urls = [ | ||||
| f"opam+{instance_url}/packages/foo/", | f"opam+{instance_url}/packages/foo/", | ||||
| ] | ] | ||||
| result_urls = [origin.url for origin in scheduler_origins] | result_urls = [origin.url for origin in scheduler_origins] | ||||
| assert expected_urls == result_urls | assert expected_urls == result_urls | ||||
| def test_opam_binary(datadir, swh_scheduler): | def test_opam_binary(datadir, swh_scheduler, tmp_path): | ||||
| instance_url = f"file://{datadir}/fake_opam_repo" | instance_url = f"file://{datadir}/fake_opam_repo" | ||||
| lister = OpamLister( | lister = OpamLister( | ||||
| swh_scheduler, | swh_scheduler, | ||||
| url=instance_url, | url=instance_url, | ||||
| instance="fake", | instance="fake", | ||||
| opam_root=mkdtemp(prefix="swh_opam_lister"), | opam_root=mkdtemp(dir=tmp_path, prefix="swh_opam_lister"), | ||||
| ) | ) | ||||
| stats = lister.run() | stats = lister.run() | ||||
| assert stats.pages == 4 | assert stats.pages == 4 | ||||
| assert stats.origins == 4 | assert stats.origins == 4 | ||||
| scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | ||||
| expected_urls = [ | expected_urls = [ | ||||
| f"opam+{instance_url}/packages/agrid/", | f"opam+{instance_url}/packages/agrid/", | ||||
| f"opam+{instance_url}/packages/calculon/", | f"opam+{instance_url}/packages/calculon/", | ||||
| f"opam+{instance_url}/packages/directories/", | f"opam+{instance_url}/packages/directories/", | ||||
| f"opam+{instance_url}/packages/ocb/", | f"opam+{instance_url}/packages/ocb/", | ||||
| ] | ] | ||||
| result_urls = [origin.url for origin in scheduler_origins] | result_urls = [origin.url for origin in scheduler_origins] | ||||
| assert expected_urls == result_urls | assert expected_urls == result_urls | ||||
| def test_opam_multi_instance(datadir, swh_scheduler): | def test_opam_multi_instance(datadir, swh_scheduler, tmp_path): | ||||
| instance_url = f"file://{datadir}/fake_opam_repo" | instance_url = f"file://{datadir}/fake_opam_repo" | ||||
| lister = OpamLister( | lister = OpamLister( | ||||
| swh_scheduler, | swh_scheduler, | ||||
| url=instance_url, | url=instance_url, | ||||
| instance="fake", | instance="fake", | ||||
| opam_root=mkdtemp(prefix="swh_opam_lister"), | opam_root=mkdtemp(dir=tmp_path, prefix="swh_opam_lister"), | ||||
| ) | ) | ||||
| stats = lister.run() | stats = lister.run() | ||||
| assert stats.pages == 4 | assert stats.pages == 4 | ||||
| assert stats.origins == 4 | assert stats.origins == 4 | ||||
| scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | ||||
| Show All 11 Lines | |||||