Page MenuHomeSoftware Heritage
Differential D4300

Install postgresql_fact fixture for faster postgres tests
ClosedPublic
Actions

Authored by ardumont on Oct 19 2020, 10:04 AM.

Details

Reviewers
vlorentz
Group Reviewers
Reviewers
Commits
rDCORE0c01f4a09284: Install postgresql_fact fixture for faster postgres tests
Summary

This moves the SWHDatabaseJanitor from swh.storage to swh.core. This also adds
tests to it.

This will allow to share the faster behavior on other swh services with
postgres backends (vault, scheduler, etc...).

Test Plan

tox

Diff Detail

Event Timeline

ardumont created this revision.Oct 19 2020, 10:04 AM
Herald added a reviewer: Reviewers. · View Herald TranscriptOct 19 2020, 10:04 AM
swh-public-ci added a comment.Oct 19 2020, 10:05 AM

Build is green

Patch application report for D4300 (id=15188)

Could not rebase; Attempt merge onto 1657fb827f...

Updating 1657fb8..4d11b30
Fast-forward
 mypy.ini                                           |   3 +
 requirements-db.txt                                |   1 +
 requirements-test-db.txt                           |   2 +-
 swh/core/api/asynchronous.py                       | 128 +++++++++++++++++---
 swh/core/api/tests/conftest.py                     |   3 +
 swh/core/api/tests/test_async.py                   |  38 +++---
 swh/core/api/tests/test_rpc_server_asynchronous.py | 132 +++++++++++++++++++++
 swh/core/db/db_utils.py                            | 104 +++++++++++++++-
 swh/core/db/pytest_plugin.py                       |  59 +++++++++
 swh/core/db/tests/conftest.py                      |   2 +
 swh/core/db/tests/data/0-schema.sql                |  13 ++
 swh/core/db/tests/data/1-data.sql                  |  10 ++
 swh/core/db/tests/test_db_utils.py                 |  56 +++++++++
 tox.ini                                            |   1 +
 14 files changed, 509 insertions(+), 43 deletions(-)
 create mode 100644 swh/core/api/tests/conftest.py
 create mode 100644 swh/core/api/tests/test_rpc_server_asynchronous.py
 create mode 100644 swh/core/db/pytest_plugin.py
 create mode 100644 swh/core/db/tests/data/0-schema.sql
 create mode 100644 swh/core/db/tests/data/1-data.sql
 create mode 100644 swh/core/db/tests/test_db_utils.py
Changes applied before test
commit 4d11b3055cfea7f9271d02dff9940c9cf13b67d5
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sun Oct 18 14:46:39 2020 +0200

    Install postgresql_fact fixture for faster postgres tests
    
    This moves the SWHDatabaseJanitor from swh.storage to swh.core. This also adds
    tests to it.
    
    This will allow to share the behavior on other swh services with postgres
    backends (vault, scheduler, etc...).

commit 604a08cbe80439b35ce0197805b6a4c536d96e62
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sat Oct 17 19:06:57 2020 +0200

    asynchronous.RPCServerApp: Align implementation with api.RPCServerApp
    
    This will allow to align the Vault's client/server code with the other rpc
    servers. And start adding one typed interface with more consistent checks in
    between implementations.

commit 7f928a07d91460efbbfbc6c5264a4fdd60e58be2
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sun Oct 18 10:25:01 2020 +0200

    api.tests.test_async: Simplify fixture setup

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/94/ for more details.

Harbormaster completed remote builds in B16425: Diff 15188.Oct 19 2020, 10:05 AM
vlorentz added a subscriber: vlorentz.Oct 19 2020, 12:36 PM
vlorentz added inline comments.
swh/core/db/db_utils.py
323

not covered, according to phab

ardumont added inline comments.Oct 19 2020, 4:10 PM
swh/core/db/db_utils.py
323

yes, i'm gonna look at how to add sequence in there.

ardumont updated this revision to Diff 15219.Oct 19 2020, 4:24 PM

Test the sequences are reset as well

swh-public-ci added a comment.Oct 19 2020, 4:25 PM

Build is green

Patch application report for D4300 (id=15219)

Rebasing onto 7f928a07d9...

Current branch diff-target is up to date.
Changes applied before test
commit 966a40142e23d8e6a258991055e65392e9acd3f7
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sun Oct 18 14:46:39 2020 +0200

    Install postgresql_fact fixture for faster postgres tests
    
    This moves the SWHDatabaseJanitor from swh.storage to swh.core. This also adds
    tests to it.
    
    This will allow to share the behavior on other swh services with postgres
    backends (vault, scheduler, etc...).

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/97/ for more details.

Harbormaster completed remote builds in B16454: Diff 15219.Oct 19 2020, 4:25 PM
ardumont updated this revision to Diff 15234.Oct 19 2020, 6:32 PM
  • Fix typo in docstring
  • Add other dostrings
swh-public-ci added a comment.Oct 19 2020, 6:33 PM

Build is green

Patch application report for D4300 (id=15234)

Rebasing onto 7f928a07d9...

Current branch diff-target is up to date.
Changes applied before test
commit 6b311c8f7794a548c68c786e5adb79dff609843a
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sun Oct 18 14:46:39 2020 +0200

    Install postgresql_fact fixture for faster postgres tests
    
    This moves the SWHDatabaseJanitor from swh.storage to swh.core. This also adds
    tests to it.
    
    This will allow to share the behavior on other swh services with postgres
    backends (vault, scheduler, etc...).

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/98/ for more details.

Harbormaster completed remote builds in B16467: Diff 15234.Oct 19 2020, 6:33 PM
ardumont mentioned this in D4310: vault.tests: Make postgresql fixture faster.Oct 19 2020, 7:05 PM
ardumont updated this revision to Diff 15287.Oct 21 2020, 6:04 PM

Add coverage and docstring

swh-public-ci added a comment.Oct 21 2020, 6:05 PM

Build is green

Patch application report for D4300 (id=15287)

Rebasing onto 7f928a07d9...

Current branch diff-target is up to date.
Changes applied before test
commit 0c01f4a09284b8843ac238f61f0a9f3ba3a0cd7a
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Sun Oct 18 14:46:39 2020 +0200

    Install postgresql_fact fixture for faster postgres tests
    
    This moves the SWHDatabaseJanitor from swh.storage to swh.core. This also adds
    tests to it.
    
    This will allow to share the behavior on other swh services with postgres
    backends (vault, scheduler, etc...).

See https://jenkins.softwareheritage.org/job/DCORE/job/tests-on-diff/101/ for more details.

Harbormaster completed remote builds in B16518: Diff 15287.Oct 21 2020, 6:05 PM
vlorentz accepted this revision.Oct 22 2020, 10:53 AM
This revision is now accepted and ready to land.Oct 22 2020, 10:53 AM
Closed by commit rDCORE0c01f4a09284: Install postgresql_fact fixture for faster postgres tests (authored by ardumont). · Explain WhyOct 22 2020, 11:09 AM
This revision was automatically updated to reflect the committed changes.
ardumont added a commit: rDCORE0c01f4a09284: Install postgresql_fact fixture for faster postgres tests.
ardumont mentioned this in D4332: requirements-swh: Bump requirements to fix webapp build.Oct 22 2020, 3:58 PM

Revision Contents
Changeset List

PathSizeCoverage (All)Coverage (Touched)
3 lines--
1 line--
2 lines--
swh/
core/
db/
104 lines35%
Loading...
59 lines78%
Loading...
tests/
2 lines100%
Loading...
data/
13 lines--
10 lines--
56 lines100%
Loading...

Diff 15188

View Options

mypy.ini

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Linesdef test_loader_svnrdump_no_such_revision(swh_storage, tmp_path, datadir):
) )
repo_url = f"file://{repo_path}" repo_url = f"file://{repo_path}"
loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path) loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
actual_visit = assert_last_visit_matches( actual_visit = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
loader2 = SvnLoaderFromRemoteDump( loader2 = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=loading_path swh_storage, repo_url, temp_directory=loading_path
) )
# Visiting a second time the same repository should be uneventful... # Visiting a second time the same repository should be uneventful...
assert loader2.load() == {"status": "uneventful"} assert loader2.load() == {"status": "uneventful"}
actual_visit2 = assert_last_visit_matches( actual_visit2 = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
Show All 16 Linesdef test_loader_svn_new_visit(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats == { assert stats == {
"content": 19, "content": 19,
"directory": 17, "directory": 17,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
Show All 18 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Linesdef test_loader_tampered_repository(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader( loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
) )
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"), snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"),
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 1 assert stats["origin"] == 1
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 2 assert stats["snapshot"] == 2
def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path): def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
Show All 14 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader( loader = SvnLoader(
swh_storage, swh_storage,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
▲ Show 20 Lines • Show All 461 Lines • ▼ Show 20 Linesdef test_loader_svn_dir_added_then_removed(swh_storage, datadir, tmp_path):
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path): def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
origin_url = f"svn://{archive_name}" origin_url = f"svn://{archive_name}"
dump_filename = f"{archive_name}.dump" dump_filename = f"{archive_name}.dump"
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 5, "directory": 5,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 3, "revision": 3,
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Linesdef test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 5, "content": 5,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Linesdef test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
paths = get_head_revision_paths_info(loader) paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed # end of lines should not have been processed
assert ( assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"]) loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content == file_content
) )
Show All 26 Lines):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
# post loading will detect an issue and make a partial visit with a snapshot # post loading will detect an issue and make a partial visit with a snapshot
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="partial", type="svn", loader.storage, repo_url, status="partial", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 2, "revision": 2,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path): def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderRevisionDivergence(SvnLoader): class SvnLoaderRevisionDivergence(SvnLoader):
def _check_revision_divergence(self, count, rev, dir_id): def _check_revision_divergence(self, count, rev, dir_id):
raise ValueError("revision divergence detected") raise ValueError("revision divergence detected")
loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="partial", status="partial",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_delete_directory_while_file_has_same_prefix( def test_loader_delete_directory_while_file_has_same_prefix(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 31 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental(swh_storage, repo_url, tmp_path): def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
( (
Show All 12 Linesdef test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
data=b"Hello World!\r\n", data=b"Hello World!\r\n",
) )
], ],
) )
# second load, incremental # second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:eol-style property on file", "Unset svn:eol-style property on file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
properties={"svn:eol-style": None}, properties={"svn:eol-style": None},
) )
], ],
) )
# third load, incremental # third load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental_replay_start_with_empty_directory( def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, repo_url, tmp_path swh_storage, mocker, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
("Add a file"), ("Add a file"),
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
) )
], ],
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n",
▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path): def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path): def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader): class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader):
Show All 16 Linesdef test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file in a directory and a svn link to the directory.", "Add an executable file in a directory and a svn link to the directory.",
Show All 30 Linesdef test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
@pytest.fixture @pytest.fixture
def external_repo_url(tmpdir_factory): def external_repo_url(tmpdir_factory):
# create a repository # create a repository
return create_repo(tmpdir_factory.mktemp("external")) return create_repo(tmpdir_factory.mktemp("external"))
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:externals property on trunk/externals path", "Unset svn:externals property on trunk/externals path",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="trunk/externals/", path="trunk/externals/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
# second load # second load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path): def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Create repository structure.", "Create repository structure.",
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_modification( def test_loader_with_valid_externals_modification(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_and_versioned_path( def test_loader_with_valid_externals_and_versioned_path(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_externals_and_versioned_path( def test_loader_with_invalid_externals_and_versioned_path(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_then_remove_and_add_as_local( def test_loader_set_externals_then_remove_and_add_as_local(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 40 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path): def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add trunk directory and set invalid external", "Add trunk directory and set invalid external",
Show All 21 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_with_versioned_file_overlap( def test_loader_set_externals_with_versioned_file_overlap(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_dump_loader_relative_externals_detection( def test_dump_loader_relative_externals_detection(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 34 Lines):
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.svnrepo.has_relative_externals assert loader.svnrepo.has_relative_externals
add_commit( add_commit(
repo_url, repo_url,
"Unset external in repository to load", "Unset external in repository to load",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="project1/", path="project1/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert not loader.svnrepo.has_relative_externals assert not loader.svnrepo.has_relative_externals
def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path): def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache
View Options

requirements-db.txt

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Linesdef test_loader_svnrdump_no_such_revision(swh_storage, tmp_path, datadir):
) )
repo_url = f"file://{repo_path}" repo_url = f"file://{repo_path}"
loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path) loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
actual_visit = assert_last_visit_matches( actual_visit = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
loader2 = SvnLoaderFromRemoteDump( loader2 = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=loading_path swh_storage, repo_url, temp_directory=loading_path
) )
# Visiting a second time the same repository should be uneventful... # Visiting a second time the same repository should be uneventful...
assert loader2.load() == {"status": "uneventful"} assert loader2.load() == {"status": "uneventful"}
actual_visit2 = assert_last_visit_matches( actual_visit2 = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
Show All 16 Linesdef test_loader_svn_new_visit(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats == { assert stats == {
"content": 19, "content": 19,
"directory": 17, "directory": 17,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
Show All 18 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Linesdef test_loader_tampered_repository(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader( loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
) )
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"), snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"),
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 1 assert stats["origin"] == 1
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 2 assert stats["snapshot"] == 2
def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path): def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
Show All 14 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader( loader = SvnLoader(
swh_storage, swh_storage,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
▲ Show 20 Lines • Show All 461 Lines • ▼ Show 20 Linesdef test_loader_svn_dir_added_then_removed(swh_storage, datadir, tmp_path):
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path): def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
origin_url = f"svn://{archive_name}" origin_url = f"svn://{archive_name}"
dump_filename = f"{archive_name}.dump" dump_filename = f"{archive_name}.dump"
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 5, "directory": 5,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 3, "revision": 3,
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Linesdef test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 5, "content": 5,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Linesdef test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
paths = get_head_revision_paths_info(loader) paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed # end of lines should not have been processed
assert ( assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"]) loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content == file_content
) )
Show All 26 Lines):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
# post loading will detect an issue and make a partial visit with a snapshot # post loading will detect an issue and make a partial visit with a snapshot
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="partial", type="svn", loader.storage, repo_url, status="partial", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 2, "revision": 2,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path): def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderRevisionDivergence(SvnLoader): class SvnLoaderRevisionDivergence(SvnLoader):
def _check_revision_divergence(self, count, rev, dir_id): def _check_revision_divergence(self, count, rev, dir_id):
raise ValueError("revision divergence detected") raise ValueError("revision divergence detected")
loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="partial", status="partial",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_delete_directory_while_file_has_same_prefix( def test_loader_delete_directory_while_file_has_same_prefix(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 31 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental(swh_storage, repo_url, tmp_path): def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
( (
Show All 12 Linesdef test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
data=b"Hello World!\r\n", data=b"Hello World!\r\n",
) )
], ],
) )
# second load, incremental # second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:eol-style property on file", "Unset svn:eol-style property on file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
properties={"svn:eol-style": None}, properties={"svn:eol-style": None},
) )
], ],
) )
# third load, incremental # third load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental_replay_start_with_empty_directory( def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, repo_url, tmp_path swh_storage, mocker, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
("Add a file"), ("Add a file"),
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
) )
], ],
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n",
▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path): def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path): def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader): class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader):
Show All 16 Linesdef test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file in a directory and a svn link to the directory.", "Add an executable file in a directory and a svn link to the directory.",
Show All 30 Linesdef test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
@pytest.fixture @pytest.fixture
def external_repo_url(tmpdir_factory): def external_repo_url(tmpdir_factory):
# create a repository # create a repository
return create_repo(tmpdir_factory.mktemp("external")) return create_repo(tmpdir_factory.mktemp("external"))
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:externals property on trunk/externals path", "Unset svn:externals property on trunk/externals path",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="trunk/externals/", path="trunk/externals/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
# second load # second load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path): def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Create repository structure.", "Create repository structure.",
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_modification( def test_loader_with_valid_externals_modification(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_and_versioned_path( def test_loader_with_valid_externals_and_versioned_path(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_externals_and_versioned_path( def test_loader_with_invalid_externals_and_versioned_path(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_then_remove_and_add_as_local( def test_loader_set_externals_then_remove_and_add_as_local(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 40 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path): def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add trunk directory and set invalid external", "Add trunk directory and set invalid external",
Show All 21 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_with_versioned_file_overlap( def test_loader_set_externals_with_versioned_file_overlap(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_dump_loader_relative_externals_detection( def test_dump_loader_relative_externals_detection(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 34 Lines):
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.svnrepo.has_relative_externals assert loader.svnrepo.has_relative_externals
add_commit( add_commit(
repo_url, repo_url,
"Unset external in repository to load", "Unset external in repository to load",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="project1/", path="project1/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert not loader.svnrepo.has_relative_externals assert not loader.svnrepo.has_relative_externals
def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path): def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache
View Options

requirements-test-db.txt

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Linesdef test_loader_svnrdump_no_such_revision(swh_storage, tmp_path, datadir):
) )
repo_url = f"file://{repo_path}" repo_url = f"file://{repo_path}"
loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path) loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
actual_visit = assert_last_visit_matches( actual_visit = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
loader2 = SvnLoaderFromRemoteDump( loader2 = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=loading_path swh_storage, repo_url, temp_directory=loading_path
) )
# Visiting a second time the same repository should be uneventful... # Visiting a second time the same repository should be uneventful...
assert loader2.load() == {"status": "uneventful"} assert loader2.load() == {"status": "uneventful"}
actual_visit2 = assert_last_visit_matches( actual_visit2 = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
Show All 16 Linesdef test_loader_svn_new_visit(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats == { assert stats == {
"content": 19, "content": 19,
"directory": 17, "directory": 17,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
Show All 18 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Linesdef test_loader_tampered_repository(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader( loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
) )
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"), snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"),
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 1 assert stats["origin"] == 1
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 2 assert stats["snapshot"] == 2
def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path): def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
Show All 14 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader( loader = SvnLoader(
swh_storage, swh_storage,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
▲ Show 20 Lines • Show All 461 Lines • ▼ Show 20 Linesdef test_loader_svn_dir_added_then_removed(swh_storage, datadir, tmp_path):
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path): def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
origin_url = f"svn://{archive_name}" origin_url = f"svn://{archive_name}"
dump_filename = f"{archive_name}.dump" dump_filename = f"{archive_name}.dump"
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 5, "directory": 5,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 3, "revision": 3,
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Linesdef test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 5, "content": 5,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Linesdef test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
paths = get_head_revision_paths_info(loader) paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed # end of lines should not have been processed
assert ( assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"]) loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content == file_content
) )
Show All 26 Lines):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
# post loading will detect an issue and make a partial visit with a snapshot # post loading will detect an issue and make a partial visit with a snapshot
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="partial", type="svn", loader.storage, repo_url, status="partial", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 2, "revision": 2,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path): def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderRevisionDivergence(SvnLoader): class SvnLoaderRevisionDivergence(SvnLoader):
def _check_revision_divergence(self, count, rev, dir_id): def _check_revision_divergence(self, count, rev, dir_id):
raise ValueError("revision divergence detected") raise ValueError("revision divergence detected")
loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="partial", status="partial",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_delete_directory_while_file_has_same_prefix( def test_loader_delete_directory_while_file_has_same_prefix(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 31 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental(swh_storage, repo_url, tmp_path): def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
( (
Show All 12 Linesdef test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
data=b"Hello World!\r\n", data=b"Hello World!\r\n",
) )
], ],
) )
# second load, incremental # second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:eol-style property on file", "Unset svn:eol-style property on file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
properties={"svn:eol-style": None}, properties={"svn:eol-style": None},
) )
], ],
) )
# third load, incremental # third load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental_replay_start_with_empty_directory( def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, repo_url, tmp_path swh_storage, mocker, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
("Add a file"), ("Add a file"),
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
) )
], ],
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n",
▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path): def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path): def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader): class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader):
Show All 16 Linesdef test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file in a directory and a svn link to the directory.", "Add an executable file in a directory and a svn link to the directory.",
Show All 30 Linesdef test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
@pytest.fixture @pytest.fixture
def external_repo_url(tmpdir_factory): def external_repo_url(tmpdir_factory):
# create a repository # create a repository
return create_repo(tmpdir_factory.mktemp("external")) return create_repo(tmpdir_factory.mktemp("external"))
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:externals property on trunk/externals path", "Unset svn:externals property on trunk/externals path",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="trunk/externals/", path="trunk/externals/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
# second load # second load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path): def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Create repository structure.", "Create repository structure.",
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_modification( def test_loader_with_valid_externals_modification(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_and_versioned_path( def test_loader_with_valid_externals_and_versioned_path(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_externals_and_versioned_path( def test_loader_with_invalid_externals_and_versioned_path(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_then_remove_and_add_as_local( def test_loader_set_externals_then_remove_and_add_as_local(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 40 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path): def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add trunk directory and set invalid external", "Add trunk directory and set invalid external",
Show All 21 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_with_versioned_file_overlap( def test_loader_set_externals_with_versioned_file_overlap(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_dump_loader_relative_externals_detection( def test_dump_loader_relative_externals_detection(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 34 Lines):
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.svnrepo.has_relative_externals assert loader.svnrepo.has_relative_externals
add_commit( add_commit(
repo_url, repo_url,
"Unset external in repository to load", "Unset external in repository to load",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="project1/", path="project1/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert not loader.svnrepo.has_relative_externals assert not loader.svnrepo.has_relative_externals
def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path): def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache
View Options

swh/core/db/db_utils.py

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Linesdef test_loader_svnrdump_no_such_revision(swh_storage, tmp_path, datadir):
) )
repo_url = f"file://{repo_path}" repo_url = f"file://{repo_path}"
loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path) loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
actual_visit = assert_last_visit_matches( actual_visit = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
loader2 = SvnLoaderFromRemoteDump( loader2 = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=loading_path swh_storage, repo_url, temp_directory=loading_path
) )
# Visiting a second time the same repository should be uneventful... # Visiting a second time the same repository should be uneventful...
assert loader2.load() == {"status": "uneventful"} assert loader2.load() == {"status": "uneventful"}
actual_visit2 = assert_last_visit_matches( actual_visit2 = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
Show All 16 Linesdef test_loader_svn_new_visit(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats == { assert stats == {
"content": 19, "content": 19,
"directory": 17, "directory": 17,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
Show All 18 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Linesdef test_loader_tampered_repository(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader( loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
) )
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"), snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"),
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 1 assert stats["origin"] == 1
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 2 assert stats["snapshot"] == 2
def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path): def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
Show All 14 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader( loader = SvnLoader(
swh_storage, swh_storage,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
▲ Show 20 Lines • Show All 461 Lines • ▼ Show 20 Linesdef test_loader_svn_dir_added_then_removed(swh_storage, datadir, tmp_path):
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path): def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
origin_url = f"svn://{archive_name}" origin_url = f"svn://{archive_name}"
dump_filename = f"{archive_name}.dump" dump_filename = f"{archive_name}.dump"
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 5, "directory": 5,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 3, "revision": 3,
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Linesdef test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 5, "content": 5,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Linesdef test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
paths = get_head_revision_paths_info(loader) paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed # end of lines should not have been processed
assert ( assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"]) loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content == file_content
) )
Show All 26 Lines):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
# post loading will detect an issue and make a partial visit with a snapshot # post loading will detect an issue and make a partial visit with a snapshot
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="partial", type="svn", loader.storage, repo_url, status="partial", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 2, "revision": 2,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path): def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderRevisionDivergence(SvnLoader): class SvnLoaderRevisionDivergence(SvnLoader):
def _check_revision_divergence(self, count, rev, dir_id): def _check_revision_divergence(self, count, rev, dir_id):
raise ValueError("revision divergence detected") raise ValueError("revision divergence detected")
loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="partial", status="partial",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_delete_directory_while_file_has_same_prefix( def test_loader_delete_directory_while_file_has_same_prefix(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 31 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental(swh_storage, repo_url, tmp_path): def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
( (
Show All 12 Linesdef test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
data=b"Hello World!\r\n", data=b"Hello World!\r\n",
) )
], ],
) )
# second load, incremental # second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:eol-style property on file", "Unset svn:eol-style property on file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
properties={"svn:eol-style": None}, properties={"svn:eol-style": None},
) )
], ],
) )
# third load, incremental # third load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental_replay_start_with_empty_directory( def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, repo_url, tmp_path swh_storage, mocker, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
("Add a file"), ("Add a file"),
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
) )
], ],
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n",
▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path): def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path): def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader): class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader):
Show All 16 Linesdef test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file in a directory and a svn link to the directory.", "Add an executable file in a directory and a svn link to the directory.",
Show All 30 Linesdef test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
@pytest.fixture @pytest.fixture
def external_repo_url(tmpdir_factory): def external_repo_url(tmpdir_factory):
# create a repository # create a repository
return create_repo(tmpdir_factory.mktemp("external")) return create_repo(tmpdir_factory.mktemp("external"))
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:externals property on trunk/externals path", "Unset svn:externals property on trunk/externals path",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="trunk/externals/", path="trunk/externals/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
# second load # second load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path): def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Create repository structure.", "Create repository structure.",
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_modification( def test_loader_with_valid_externals_modification(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_and_versioned_path( def test_loader_with_valid_externals_and_versioned_path(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_externals_and_versioned_path( def test_loader_with_invalid_externals_and_versioned_path(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_then_remove_and_add_as_local( def test_loader_set_externals_then_remove_and_add_as_local(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 40 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path): def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add trunk directory and set invalid external", "Add trunk directory and set invalid external",
Show All 21 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_with_versioned_file_overlap( def test_loader_set_externals_with_versioned_file_overlap(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_dump_loader_relative_externals_detection( def test_dump_loader_relative_externals_detection(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 34 Lines):
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.svnrepo.has_relative_externals assert loader.svnrepo.has_relative_externals
add_commit( add_commit(
repo_url, repo_url,
"Unset external in repository to load", "Unset external in repository to load",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="project1/", path="project1/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert not loader.svnrepo.has_relative_externals assert not loader.svnrepo.has_relative_externals
def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path): def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache
View Options

swh/core/db/pytest_plugin.py

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Linesdef test_loader_svnrdump_no_such_revision(swh_storage, tmp_path, datadir):
) )
repo_url = f"file://{repo_path}" repo_url = f"file://{repo_path}"
loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path) loader = SvnLoaderFromRemoteDump(swh_storage, repo_url, temp_directory=loading_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
actual_visit = assert_last_visit_matches( actual_visit = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
loader2 = SvnLoaderFromRemoteDump( loader2 = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=loading_path swh_storage, repo_url, temp_directory=loading_path
) )
# Visiting a second time the same repository should be uneventful... # Visiting a second time the same repository should be uneventful...
assert loader2.load() == {"status": "uneventful"} assert loader2.load() == {"status": "uneventful"}
actual_visit2 = assert_last_visit_matches( actual_visit2 = assert_last_visit_matches(
swh_storage, repo_url, status="full", type="svn", swh_storage, repo_url, status="full", type="svn",
Show All 16 Linesdef test_loader_svn_new_visit(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats == { assert stats == {
"content": 19, "content": 19,
"directory": 17, "directory": 17,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
Show All 18 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
▲ Show 20 Lines • Show All 56 Lines • ▼ Show 20 Linesdef test_loader_tampered_repository(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(loader.snapshot, loader.storage)
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader( loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
) )
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"), snapshot=hash_to_bytes("5aa61959e788e281fd6e187053d0f46c68e8d8bb"),
) )
check_snapshot(loader.snapshot, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 1 assert stats["origin"] == 1
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 2 assert stats["snapshot"] == 2
def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path): def test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
Show All 14 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader( loader = SvnLoader(
swh_storage, swh_storage,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
▲ Show 20 Lines • Show All 461 Lines • ▼ Show 20 Linesdef test_loader_svn_dir_added_then_removed(swh_storage, datadir, tmp_path):
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path): def test_loader_svn_loader_from_dump_archive(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
origin_url = f"svn://{archive_name}" origin_url = f"svn://{archive_name}"
dump_filename = f"{archive_name}.dump" dump_filename = f"{archive_name}.dump"
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 5, "directory": 5,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 3, "revision": 3,
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Linesdef test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# check loaded objects are those expected # check loaded objects are those expected
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 5, "content": 5,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Linesdef test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
paths = get_head_revision_paths_info(loader) paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed # end of lines should not have been processed
assert ( assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"]) loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content == file_content
) )
Show All 26 Lines):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
# post loading will detect an issue and make a partial visit with a snapshot # post loading will detect an issue and make a partial visit with a snapshot
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="partial", type="svn", loader.storage, repo_url, status="partial", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert get_stats(loader.storage) == { assert get_stats(loader.storage) == {
"content": 2, "content": 2,
"directory": 2, "directory": 2,
"origin": 1, "origin": 1,
"origin_visit": 1, "origin_visit": 1,
"release": 0, "release": 0,
"revision": 2, "revision": 2,
▲ Show 20 Lines • Show All 71 Lines • ▼ Show 20 Linesdef test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path): def test_loader_last_revision_divergence(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderRevisionDivergence(SvnLoader): class SvnLoaderRevisionDivergence(SvnLoader):
def _check_revision_divergence(self, count, rev, dir_id): def _check_revision_divergence(self, count, rev, dir_id):
raise ValueError("revision divergence detected") raise ValueError("revision divergence detected")
loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoaderRevisionDivergence(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "failed"} assert loader.load() == {"status": "failed"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="partial", status="partial",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_delete_directory_while_file_has_same_prefix( def test_loader_delete_directory_while_file_has_same_prefix(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 31 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental(swh_storage, repo_url, tmp_path): def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
( (
Show All 12 Linesdef test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
data=b"Hello World!\r\n", data=b"Hello World!\r\n",
) )
], ],
) )
# second load, incremental # second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:eol-style property on file", "Unset svn:eol-style property on file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="file_with_crlf_eol.txt", path="file_with_crlf_eol.txt",
properties={"svn:eol-style": None}, properties={"svn:eol-style": None},
) )
], ],
) )
# third load, incremental # third load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_svn_loader_incremental_replay_start_with_empty_directory( def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, repo_url, tmp_path swh_storage, mocker, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
("Add a file"), ("Add a file"),
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
) )
], ],
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# second commit # second commit
add_commit( add_commit(
repo_url, repo_url,
"Modify previously added file", "Modify previously added file",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n", change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"bar\n",
▲ Show 20 Lines • Show All 75 Lines • ▼ Show 20 Lines):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path): def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file and a svn link to it.", "Add an executable file and a svn link to it.",
Show All 30 Linesdef test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path): def test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader): class SvnLoaderPostLoadLocalDirIsEmpty(SvnLoader):
Show All 16 Linesdef test_loader_svn_empty_local_dir_before_post_load(swh_storage, datadir, tmp_path):
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage)
def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path): def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add an executable file in a directory and a svn link to the directory.", "Add an executable file in a directory and a svn link to the directory.",
Show All 30 Linesdef test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# the repository filesystem it reconstructed does not differ from a subversion # the repository filesystem it reconstructed does not differ from a subversion
# export of that revision # export of that revision
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
@pytest.fixture @pytest.fixture
def external_repo_url(tmpdir_factory): def external_repo_url(tmpdir_factory):
# create a repository # create a repository
return create_repo(tmpdir_factory.mktemp("external")) return create_repo(tmpdir_factory.mktemp("external"))
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines):
) )
# first load # first load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
# third commit # third commit
add_commit( add_commit(
repo_url, repo_url,
"Unset svn:externals property on trunk/externals path", "Unset svn:externals property on trunk/externals path",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="trunk/externals/", path="trunk/externals/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
# second load # second load
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path): def test_loader_with_invalid_svn_externals(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Create repository structure.", "Create repository structure.",
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_modification( def test_loader_with_valid_externals_modification(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_valid_externals_and_versioned_path( def test_loader_with_valid_externals_and_versioned_path(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_with_invalid_externals_and_versioned_path( def test_loader_with_invalid_externals_and_versioned_path(
swh_storage, repo_url, tmp_path swh_storage, repo_url, tmp_path
): ):
# first commit # first commit
add_commit( add_commit(
Show All 25 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_then_remove_and_add_as_local( def test_loader_set_externals_then_remove_and_add_as_local(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 40 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path): def test_loader_set_invalid_externals_then_remove(swh_storage, repo_url, tmp_path):
# first commit # first commit
add_commit( add_commit(
repo_url, repo_url,
"Add trunk directory and set invalid external", "Add trunk directory and set invalid external",
Show All 21 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_loader_set_externals_with_versioned_file_overlap( def test_loader_set_externals_with_versioned_file_overlap(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
def test_dump_loader_relative_externals_detection( def test_dump_loader_relative_externals_detection(
swh_storage, repo_url, external_repo_url, tmp_path swh_storage, repo_url, external_repo_url, tmp_path
): ):
add_commit( add_commit(
external_repo_url, external_repo_url,
Show All 34 Lines):
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert loader.svnrepo.has_relative_externals assert loader.svnrepo.has_relative_externals
add_commit( add_commit(
repo_url, repo_url,
"Unset external in repository to load", "Unset external in repository to load",
[ [
CommitChange( CommitChange(
change_type=CommitChangeType.AddOrUpdate, change_type=CommitChangeType.AddOrUpdate,
path="project1/", path="project1/",
properties={"svn:externals": None}, properties={"svn:externals": None},
), ),
], ],
) )
loader = SvnLoaderFromRemoteDump( loader = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path, check_revision=1 swh_storage, repo_url, temp_directory=tmp_path, check_revision=1
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert not loader.svnrepo.has_relative_externals assert not loader.svnrepo.has_relative_externals
def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path): def test_loader_externals_cache(swh_storage, repo_url, external_repo_url, tmp_path):
# first commit on external # first commit on external
add_commit( add_commit(
external_repo_url, external_repo_url,
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines add_commit(
], ],
) )
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn", loader.storage, repo_url, status="full", type="svn",
) )
check_snapshot(loader.snapshot, loader.storage)
assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache assert (external_url, None) in loader.svnrepo.swhreplay.editor.externals_cache
View Options

swh/core/db/tests/conftest.py

Loading...
View Options

swh/core/db/tests/data/0-schema.sql

Loading...
View Options

swh/core/db/tests/data/1-data.sql

Loading...
View Options

swh/core/db/tests/test_db_utils.py

Loading...
About · Developer information · Legal