Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/cvs/cvsclient.py
| # Copyright (C) 2015-2022 The Software Heritage developers | # Copyright (C) 2015-2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| """Minimal CVS client implementation | """Minimal CVS client implementation | ||||
| """ | """ | ||||
| import os.path | import os.path | ||||
| import socket | import socket | ||||
| import subprocess | import subprocess | ||||
| import tempfile | import tempfile | ||||
| from tenacity import retry | |||||
| from tenacity.retry import retry_if_exception_type | |||||
| from tenacity.stop import stop_after_attempt | |||||
| from swh.loader.exception import NotFound | from swh.loader.exception import NotFound | ||||
| CVS_PSERVER_PORT = 2401 | CVS_PSERVER_PORT = 2401 | ||||
| CVS_PROTOCOL_BUFFER_SIZE = 8192 | CVS_PROTOCOL_BUFFER_SIZE = 8192 | ||||
| EXAMPLE_PSERVER_URL = "pserver://user:password@cvs.example.com/cvsroot/repository" | EXAMPLE_PSERVER_URL = "pserver://user:password@cvs.example.com/cvsroot/repository" | ||||
| EXAMPLE_SSH_URL = "ssh://user@cvs.example.com/cvsroot/repository" | EXAMPLE_SSH_URL = "ssh://user@cvs.example.com/cvsroot/repository" | ||||
| VALID_RESPONSES = [ | VALID_RESPONSES = [ | ||||
| ▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | def scramble_password(password): | ||||
| return "".join(s) | return "".join(s) | ||||
| class CVSProtocolError(Exception): | class CVSProtocolError(Exception): | ||||
| pass | pass | ||||
| class CVSClient: | class CVSClient: | ||||
| # connection to an existing pserver might sometimes fail, | |||||
| # retrying the operation usually fixes the issue | |||||
| @retry( | |||||
| retry=retry_if_exception_type(NotFound), | |||||
| stop=stop_after_attempt(max_attempt_number=3), | |||||
| reraise=True, | |||||
| ) | |||||
| def connect_pserver(self, hostname, port, username, password): | def connect_pserver(self, hostname, port, username, password): | ||||
| if port is None: | if port is None: | ||||
| port = CVS_PSERVER_PORT | port = CVS_PSERVER_PORT | ||||
| if username is None: | if username is None: | ||||
| raise NotFound( | raise NotFound( | ||||
| "Username is required for " | "Username is required for " | ||||
| "a pserver connection: %s" % EXAMPLE_PSERVER_URL | "a pserver connection: %s" % EXAMPLE_PSERVER_URL | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 337 Lines • Show Last 20 Lines | |||||