Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/archive/tests/test_archive.py
| Show All 16 Lines | |||||
| URL = "https://ftp.gnu.org/gnu/8sync/" | URL = "https://ftp.gnu.org/gnu/8sync/" | ||||
| GNU_ARTIFACTS = [ | GNU_ARTIFACTS = [ | ||||
| { | { | ||||
| "time": 944729610, | "time": 944729610, | ||||
| "url": "https://ftp.gnu.org/gnu/8sync/8sync-0.1.0.tar.gz", | "url": "https://ftp.gnu.org/gnu/8sync/8sync-0.1.0.tar.gz", | ||||
| "length": 221837, | "length": 221837, | ||||
| "filename": "8sync-0.1.0.tar.gz", | "filename": "8sync-0.1.0.tar.gz", | ||||
| "version": "0.1.0", | "version": "0.1.0", | ||||
| } | }, | ||||
| { | |||||
| "time": 1480991830, | |||||
| "url": "https://ftp.gnu.org/gnu/8sync/8sync-0.2.0.tar.gz", | |||||
| "length": 238466, | |||||
| "filename": "8sync-0.2.0.tar.gz", | |||||
| "version": "0.2.0", | |||||
| }, | |||||
| ] | ] | ||||
| _expected_new_contents_first_visit = [ | _expected_new_contents_first_visit = [ | ||||
| "e9258d81faf5881a2f96a77ba609396f82cb97ad", | "e9258d81faf5881a2f96a77ba609396f82cb97ad", | ||||
| "1170cf105b04b7e2822a0e09d2acf71da7b9a130", | "1170cf105b04b7e2822a0e09d2acf71da7b9a130", | ||||
| "fbd27c3f41f2668624ffc80b7ba5db9b92ff27ac", | "fbd27c3f41f2668624ffc80b7ba5db9b92ff27ac", | ||||
| "0057bec9b5422aff9256af240b177ac0e3ac2608", | "0057bec9b5422aff9256af240b177ac0e3ac2608", | ||||
| "2b8d0d0b43a1078fc708930c8ddc2956a86c566e", | "2b8d0d0b43a1078fc708930c8ddc2956a86c566e", | ||||
| ▲ Show 20 Lines • Show All 76 Lines • ▼ Show 20 Lines | |||||
| def test_archive_visit_with_release_artifact_no_prior_visit( | def test_archive_visit_with_release_artifact_no_prior_visit( | ||||
| swh_storage, requests_mock_datadir | swh_storage, requests_mock_datadir | ||||
| ): | ): | ||||
| """With no prior visit, load a gnu project ends up with 1 snapshot | """With no prior visit, load a gnu project ends up with 1 snapshot | ||||
| """ | """ | ||||
| loader = ArchiveLoader(swh_storage, URL, artifacts=GNU_ARTIFACTS) | loader = ArchiveLoader(swh_storage, URL, artifacts=GNU_ARTIFACTS[:1]) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | assert actual_load_status["status"] == "eventful" | ||||
| expected_snapshot_first_visit_id = hash_to_bytes( | expected_snapshot_first_visit_id = hash_to_bytes( | ||||
| "c419397fd912039825ebdbea378bc6283f006bf5" | "c419397fd912039825ebdbea378bc6283f006bf5" | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | ): | ||||
| check_snapshot(expected_snapshot, swh_storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| def test_archive_2_visits_without_change(swh_storage, requests_mock_datadir): | def test_archive_2_visits_without_change(swh_storage, requests_mock_datadir): | ||||
| """With no prior visit, load a gnu project ends up with 1 snapshot | """With no prior visit, load a gnu project ends up with 1 snapshot | ||||
| """ | """ | ||||
| url = URL | url = URL | ||||
| loader = ArchiveLoader(swh_storage, url, artifacts=GNU_ARTIFACTS) | loader = ArchiveLoader(swh_storage, url, artifacts=GNU_ARTIFACTS[:1]) | ||||
| actual_load_status = loader.load() | actual_load_status = loader.load() | ||||
| assert actual_load_status["status"] == "eventful" | assert actual_load_status["status"] == "eventful" | ||||
| assert actual_load_status["snapshot_id"] is not None | assert actual_load_status["snapshot_id"] is not None | ||||
| assert_last_visit_matches(swh_storage, url, status="full", type="tar") | assert_last_visit_matches(swh_storage, url, status="full", type="tar") | ||||
| actual_load_status2 = loader.load() | actual_load_status2 = loader.load() | ||||
| Show All 39 Lines | def test_archive_2_visits_with_new_artifact(swh_storage, requests_mock_datadir): | ||||
| urls = [ | urls = [ | ||||
| m.url | m.url | ||||
| for m in requests_mock_datadir.request_history | for m in requests_mock_datadir.request_history | ||||
| if m.url.startswith("https://ftp.gnu.org") | if m.url.startswith("https://ftp.gnu.org") | ||||
| ] | ] | ||||
| assert len(urls) == 1 | assert len(urls) == 1 | ||||
| artifact2 = { | artifact2 = GNU_ARTIFACTS[1] | ||||
| "time": 1480991830, | |||||
| "url": "https://ftp.gnu.org/gnu/8sync/8sync-0.2.0.tar.gz", | |||||
| "length": 238466, | |||||
| "filename": "8sync-0.2.0.tar.gz", | |||||
| "version": "0.2.0", | |||||
| } | |||||
| loader2 = ArchiveLoader(swh_storage, url, [artifact1, artifact2]) | loader2 = ArchiveLoader(swh_storage, url, [artifact1, artifact2]) | ||||
| stats2 = get_stats(swh_storage) | stats2 = get_stats(swh_storage) | ||||
| assert stats == stats2 # ensure we share the storage | assert stats == stats2 # ensure we share the storage | ||||
| actual_load_status2 = loader2.load() | actual_load_status2 = loader2.load() | ||||
| assert actual_load_status2["status"] == "eventful" | assert actual_load_status2["status"] == "eventful" | ||||
| assert actual_load_status2["snapshot_id"] is not None | assert actual_load_status2["snapshot_id"] is not None | ||||
| ▲ Show 20 Lines • Show All 89 Lines • ▼ Show 20 Lines | ]: | ||||
| actual_id = p_info.extid(manifest_format=manifest_format) | actual_id = p_info.extid(manifest_format=manifest_format) | ||||
| assert actual_id == ( | assert actual_id == ( | ||||
| "package-manifest-sha256", | "package-manifest-sha256", | ||||
| hashlib.sha256(expected_manifest).digest(), | hashlib.sha256(expected_manifest).digest(), | ||||
| ) | ) | ||||
| with pytest.raises(KeyError): | with pytest.raises(KeyError): | ||||
| p_info.extid(manifest_format=string.Template("$a $unknown_key")) | p_info.extid(manifest_format=string.Template("$a $unknown_key")) | ||||
| def test_archive_snapshot_append(swh_storage, requests_mock_datadir): | |||||
| # first loading with a first artifact | |||||
| artifact1 = GNU_ARTIFACTS[0] | |||||
| loader = ArchiveLoader(swh_storage, URL, [artifact1], snapshot_append=True) | |||||
| actual_load_status = loader.load() | |||||
| assert actual_load_status["status"] == "eventful" | |||||
| assert actual_load_status["snapshot_id"] is not None | |||||
| assert_last_visit_matches(swh_storage, URL, status="full", type="tar") | |||||
| # check expected snapshot | |||||
| snapshot = loader.last_snapshot() | |||||
| assert len(snapshot.branches) == 2 | |||||
| branch_artifact1_name = f"releases/{artifact1['version']}".encode() | |||||
| assert b"HEAD" in snapshot.branches | |||||
| assert branch_artifact1_name in snapshot.branches | |||||
| assert snapshot.branches[b"HEAD"].target == branch_artifact1_name | |||||
| # second loading with a second artifact | |||||
| artifact2 = GNU_ARTIFACTS[1] | |||||
| loader = ArchiveLoader(swh_storage, URL, [artifact2], snapshot_append=True) | |||||
| actual_load_status = loader.load() | |||||
| assert actual_load_status["status"] == "eventful" | |||||
| assert actual_load_status["snapshot_id"] is not None | |||||
| assert_last_visit_matches(swh_storage, URL, status="full", type="tar") | |||||
| # check expected snapshot, should contain a new branch and the | |||||
| # branch for the first artifact | |||||
| snapshot = loader.last_snapshot() | |||||
| assert len(snapshot.branches) == 3 | |||||
| branch_artifact2_name = f"releases/{artifact2['version']}".encode() | |||||
| assert b"HEAD" in snapshot.branches | |||||
| assert branch_artifact2_name in snapshot.branches | |||||
| assert branch_artifact1_name in snapshot.branches | |||||
| assert snapshot.branches[b"HEAD"].target == branch_artifact2_name | |||||
| def test_archive_snapshot_append_branch_override(swh_storage, requests_mock_datadir): | |||||
| # first loading for a first artifact | |||||
| artifact1 = GNU_ARTIFACTS[0] | |||||
| loader = ArchiveLoader(swh_storage, URL, [artifact1], snapshot_append=True) | |||||
| actual_load_status = loader.load() | |||||
| assert actual_load_status["status"] == "eventful" | |||||
| assert actual_load_status["snapshot_id"] is not None | |||||
| assert_last_visit_matches(swh_storage, URL, status="full", type="tar") | |||||
| # check expected snapshot | |||||
| snapshot = loader.last_snapshot() | |||||
| assert len(snapshot.branches) == 2 | |||||
| branch_artifact1_name = f"releases/{artifact1['version']}".encode() | |||||
| assert branch_artifact1_name in snapshot.branches | |||||
| branch_target_first_visit = snapshot.branches[branch_artifact1_name].target | |||||
| # second loading for a second artifact with same version as the first one | |||||
| # but with different tarball content | |||||
| artifact2 = dict(GNU_ARTIFACTS[0]) | |||||
| artifact2["url"] = GNU_ARTIFACTS[1]["url"] | |||||
| artifact2["time"] = GNU_ARTIFACTS[1]["time"] | |||||
| artifact2["length"] = GNU_ARTIFACTS[1]["length"] | |||||
| loader = ArchiveLoader(swh_storage, URL, [artifact2], snapshot_append=True) | |||||
| actual_load_status = loader.load() | |||||
| assert actual_load_status["status"] == "eventful" | |||||
| assert actual_load_status["snapshot_id"] is not None | |||||
| assert_last_visit_matches(swh_storage, URL, status="full", type="tar") | |||||
| # check expected snapshot, should contain the same branch as previously | |||||
| # but with different target | |||||
| snapshot = loader.last_snapshot() | |||||
| assert len(snapshot.branches) == 2 | |||||
| assert branch_artifact1_name in snapshot.branches | |||||
| branch_target_second_visit = snapshot.branches[branch_artifact1_name].target | |||||
| assert branch_target_first_visit != branch_target_second_visit | |||||