Page MenuHomeSoftware Heritage
Differential D5194

Expose a pytest plugin for swh.auth.keycloak
ClosedPublic
Actions

Authored by ardumont on Mar 3 2021, 6:50 PM.

Details

Reviewers
anlambert
Group Reviewers
Reviewers
Maniphest Tasks
T3079: Boostrap swh-auth module for authentication
Commits
rDAUTH63b28edb1f40: Expose a pytest plugin for swh.auth.keycloak
Summary

This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)

Related to T3079

Diff Detail

Repository
rDAUTH Common authentication libraries
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

ardumont created this revision.Mar 3 2021, 6:50 PM
Herald added a reviewer: Reviewers. · View Herald TranscriptMar 3 2021, 6:50 PM
swh-public-ci added a comment.Mar 3 2021, 6:50 PM

Build is green

Patch application report for D5194 (id=18579)

Could not rebase; Attempt merge onto a750b58718...

Updating a750b58..9d8e662
Fast-forward
 MANIFEST.in                   |   1 +
 conftest.py                   |   6 +
 swh/auth/pytest_plugin.py     | 160 ++++++++++++++++++++++++
 swh/auth/sample_data.py       | 144 +++++++++++++++++++++
 swh/auth/tests/conftest.py    |  39 ------
 swh/auth/tests/sample_data.py | 284 ------------------------------------------
 swh/auth/tests/test_auth.py   |  86 +++++++------
 7 files changed, 353 insertions(+), 367 deletions(-)
 create mode 100644 conftest.py
 create mode 100644 swh/auth/pytest_plugin.py
 create mode 100644 swh/auth/sample_data.py
 delete mode 100644 swh/auth/tests/conftest.py
 delete mode 100644 swh/auth/tests/sample_data.py
Changes applied before test
commit 9d8e662302a24a754cbc180d38dfe622a1a42f8f
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth
    
    This will be required for module depending on it

commit 2d3a37d54c4eff3c1421f2e9b093cf1c6118cdb5
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 17:41:29 2021 +0100

    auth: Rework tests logic and add coverage to decode_token endpoint
    
    This reworks the tests logic as well to reuse the way web tests are written.
    
    This is a first step to actually use the mock class defined here as fixture for future
    modules which will depend on swh-auth (swh-web-client, swh-web and swh-deposit).
    
    Related to T3079

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

Harbormaster completed remote builds in B19669: Diff 18579.Mar 3 2021, 6:50 PM
ardumont requested review of this revision.Mar 3 2021, 6:50 PM
anlambert added a subscriber: anlambert.Mar 3 2021, 7:14 PM

I think you should leave the sample_data module inside the tests folder, create a keycloak_mock.py module (also in tests) containing the Keycloak mocked class and simply declare the keycloak_mock fixture in the pytest_plugin file.

ardumont added a comment.Mar 4 2021, 9:19 AM

I think you should leave the sample_data module inside the tests folder, create a keycloak_mock.py module (also in tests) containing the Keycloak mocked class and simply declare the keycloak_mock fixture in the pytest_plugin file.

makes more sense than the current proposed diff, thanks

ardumont added a comment.Mar 4 2021, 9:51 AM
In D5194#131506, @ardumont wrote:

I think you should leave the sample_data module inside the tests folder, create a keycloak_mock.py module (also in tests) containing the Keycloak mocked class and simply declare the keycloak_mock fixture in the pytest_plugin file.

makes more sense than the current proposed diff, thanks

mmm, except we have some hard-coded sample like USER_INFO, OIDC_PROFILE, ...

ardumont updated this revision to Diff 18584.Mar 4 2021, 10:19 AM

Rework according to review

This:

  • makes the factory even more generic.
  • instantiates the fixture we want depending on sample data (which is defined only in swh.auth.tests only)
swh-public-ci added a comment.Mar 4 2021, 10:20 AM

Build is green

Patch application report for D5194 (id=18584)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit 16d3e0ae12117e5b4a1e02baedc6781c15b33c57
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19675: Diff 18584.Mar 4 2021, 10:20 AM
ardumont added inline comments.Mar 4 2021, 10:24 AM
swh/auth/tests/sample_data.py
142

unused ^ so i dropped it.

swh/auth/tests/test_keycloak.py
30

i could also define this in ./conftest.py ^
But i thought to leave it here so it avoids one more indirection (sample_data got split up already [1])

[1] While it's more readable that way, it's still one more jump to do when on actually want to check up those variables ¯\_(ツ)_/¯.

ardumont retitled this revision from Expose a pytest plugin for swh.auth to Expose a pytest plugin for swh.auth.keycloak.Mar 4 2021, 10:25 AM
ardumont edited the summary of this revision. (Show Details)
ardumont added a task: T3079: Boostrap swh-auth module for authentication.
ardumont updated this revision to Diff 18585.Mar 4 2021, 10:33 AM
ardumont edited the summary of this revision. (Show Details)

Update docstring on attributes missing explanation.

Fix sphinx typo :py:data: for module attribute [1]

[1] https://www.sphinx-doc.org/en/master/usage/restructuredtext/domains.html#python-roles

swh-public-ci added a comment.Mar 4 2021, 10:34 AM

Build is green

Patch application report for D5194 (id=18585)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit e07dab6095ff8c6473201c1994914e69e7af55a4
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19676: Diff 18585.Mar 4 2021, 10:34 AM
anlambert requested changes to this revision.Mar 4 2021, 12:23 PM

See inline comment.

swh/auth/pytest_plugin.py
46–48

I am not a big fan to provide those data as parameter of the Keycloak mock as it will complicate the writing of tests in client modules.

I think it will be simpler to override the JWT data that Keycloak returns directly in the decode_token method. For instance, you can override the client_id value in the adequate fields of the decoded token.

Nevertheless we can keep those parameters for flexibility in tests implementation but at least they should have default values taken from the sample_data module.

This revision now requires changes to proceed.Mar 4 2021, 12:23 PM
ardumont added inline comments.Mar 4 2021, 12:40 PM
swh/auth/pytest_plugin.py
46–48

I am not a big fan to provide those data as parameter of the Keycloak mock as it will complicate the writing of tests in client modules.

not really if one defines the mock instance they wants once
and use that all around in their tests?

I think it will be simpler to override the JWT data that Keycloak returns directly in the decode_token method. For instance, you can override the client_id value in the adequate fields of the decoded token.

erf, "a pas compris" ¯\_(ツ)_/¯

I did not get it (yet?)

Nevertheless we can keep those parameters for flexibility in tests implementation but at least they should have default values taken from the sample_data module

where should be the sample-data defined then?
(It's currently in test to not expose it in the main swh.auth tree)

ardumont updated this revision to Diff 18601.Mar 4 2021, 3:31 PM

Adapt according to review/discussion

Allow re-definition of oidc_profile, user_info, raw_realm_public_key but let there be a
default from sample data.

When needed, the mock can be adapted so the decode_token method does the right and
expected thing for the tester.

swh-public-ci added a comment.Mar 4 2021, 3:32 PM

Build is green

Patch application report for D5194 (id=18601)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit bbd8b85e21ae94275ca678396ca31a4cbd098159
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19693: Diff 18601.Mar 4 2021, 3:32 PM
ardumont updated this revision to Diff 18602.Mar 4 2021, 3:34 PM

Add missing type on user groups and permissions

swh-public-ci added a comment.Mar 4 2021, 3:34 PM

Build is green

Patch application report for D5194 (id=18602)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit 13c984d7089c6300d89e16259097f72d5ea738d2
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19694: Diff 18602.Mar 4 2021, 3:34 PM
anlambert requested changes to this revision.Mar 4 2021, 3:44 PM

Just a couple of small changes to add and we should be good here.

swh/auth/pytest_plugin.py
27–35

You should also add some setter methods for those values.

In a same manner than with the set_auth_success one, client of the fixture
will simply call these methods at the beginning of their test to override
groups, permissions or token expiration date..

108

the azp field must also be set to self.client_id

swh/auth/tests/sample_data.py
7–8

I think you can safely change the value to "client-id".

This revision now requires changes to proceed.Mar 4 2021, 3:44 PM
ardumont marked an inline comment as done.Mar 4 2021, 5:04 PM
ardumont added inline comments.
swh/auth/pytest_plugin.py
27–35

I initially agreed, although i was unsure at how many setters you proposed (all, only a subset?).

Now, i'm wondering if it's not enough that callers can simply use:

instance.user_groups = wanted_user_groups
instance.user_permissions = ...

That sounds pythonic enough ^, what do you think?

46–48

agreed and adapted according to antoine's suggestion btw

108

good catch

swh/auth/tests/sample_data.py
7–8

I thought so but the decode-token part is complaining about invalid audience

Most likely the:

super().decode_token(token, options)

call which ends up with:

jose.exceptions.JWTClaimsError: Invalid audience
ardumont added inline comments.Mar 4 2021, 5:07 PM
swh/auth/tests/sample_data.py
7–8

data from decoded-token:

claims = {'acr': '1', 'allowed-origins': ['*'], 'aud': ['swh-web', 'account'], 'auth_time': 1582723100, ...}, audience = 'client-id'
ardumont added inline comments.Mar 4 2021, 5:16 PM
swh/auth/tests/sample_data.py
7–8

patching it to bypass that step works.

ardumont updated this revision to Diff 18610.Mar 4 2021, 5:21 PM

Adapt according to suggestions:

  • generic client id (means we add to patch a bit further the decoding-token step)
  • fix azp key entry

I did not modify the setters yet.

swh-public-ci added a comment.Mar 4 2021, 5:22 PM

Build is green

Patch application report for D5194 (id=18610)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit 31a3a4fc9e25ee823aeb41bd76cdc4957d286ec3
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19702: Diff 18610.Mar 4 2021, 5:22 PM
anlambert requested changes to this revision.Mar 4 2021, 5:26 PM

A small little change and we are good.

swh/auth/pytest_plugin.py
27–35

Fair enough, this is equivalent indeed. Using methods could have allowed to pass datetime objects for the expiration date but not a big deal.

99–100

Use option["verify_aud"] = False instead

swh/auth/tests/sample_data.py
7–8

I also think we can bypass the audience check as the expiration date one by setting the verify_aud option for the decode_token method to False

This revision now requires changes to proceed.Mar 4 2021, 5:26 PM
ardumont added inline comments.Mar 4 2021, 5:32 PM
swh/auth/pytest_plugin.py
99–100

\o/
way better, thanks.

ardumont added inline comments.Mar 4 2021, 5:34 PM
swh/auth/pytest_plugin.py
99–100

also "d'oh", that's what just before my eyes for the expiration check...

ardumont updated this revision to Diff 18611.Mar 4 2021, 5:36 PM

Adapt according to last and way better implementation suggestion about the check
audience step (thanks!)

swh-public-ci added a comment.Mar 4 2021, 5:37 PM

Build is green

Patch application report for D5194 (id=18611)

Rebasing onto d2f211feb9...

Current branch diff-target is up to date.
Changes applied before test
commit 63b28edb1f40ddcfe16859156ee5c61ad9bc834b
Author: Antoine R. Dumont (@ardumont) <antoine.romain.dumont@gmail.com>
Date:   Wed Mar 3 18:47:31 2021 +0100

    Expose a pytest plugin for swh.auth.keycloak
    
    This will be required for modules depending on it (swh.web, swh.web.client, swh.deposit)
    
    Related to T3079

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

Harbormaster completed remote builds in B19703: Diff 18611.Mar 4 2021, 5:37 PM
anlambert accepted this revision.Mar 4 2021, 5:38 PM
This revision is now accepted and ready to land.Mar 4 2021, 5:38 PM
Closed by commit rDAUTH63b28edb1f40: Expose a pytest plugin for swh.auth.keycloak (authored by ardumont). · Explain WhyMar 4 2021, 5:41 PM
This revision was automatically updated to reflect the committed changes.
ardumont added a commit: rDAUTH63b28edb1f40: Expose a pytest plugin for swh.auth.keycloak.

Revision Contents
Changeset List

PathSize
1 line
6 lines
swh/
auth/
pytest_plugin.py/
tests/
conftest.py
77 lines
tests/
12 lines
19 lines

Diff 18612

View Options

MANIFEST.in

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

conftest.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/auth/pytest_plugin.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/auth/tests/conftest.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/auth/tests/sample_data.py

Loading...
View Options

swh/auth/tests/test_keycloak.py

Loading...
About · Developer information · Legal