diff --git a/morgan/__init__.py b/morgan/__init__.py index bb72c79..e3b8a36 100644 --- a/morgan/__init__.py +++ b/morgan/__init__.py @@ -3,7 +3,6 @@ import argparse import configparser import hashlib -import json import os import os.path import re @@ -25,6 +24,7 @@ from morgan.__about__ import __version__ from morgan.utils import ( Cache, + Index, ListExtendingOrderedDict, is_requirement_relevant, to_single_dash, @@ -60,6 +60,7 @@ def __init__(self, args: argparse.Namespace): dict_type=ListExtendingOrderedDict, ) self.config.read(args.config) + self.index = Index(args) self.envs = {} self._supported_pyversions = [] self._supported_platforms = [] @@ -158,24 +159,7 @@ def _mirror( # noqa: C901, PLR0912 else: print(f"{requirement}") - data: dict | None = None - - # get information about this package from the Simple API in JSON - # format as per PEP 691 - request = urllib.request.Request( # noqa: S310 - f"{self.index_url}{requirement.name}/", - headers={ - "Accept": "application/vnd.pypi.simple.v1+json", - }, - ) - - response_url = "" - with urllib.request.urlopen(request) as response: # noqa: S310 - data = json.load(response) - response_url = str(response.url) - if not data: - msg = f"Failed loading metadata: {response}" - raise RuntimeError(msg) + data, response_url = self.index.get(requirement.name) # check metadata version ~1.0 v_str = data["meta"]["api-version"] @@ -449,7 +433,7 @@ def _matches_environments( # noqa: C901, PLR0912 if fileinfo.get("tags"): # At least one of the tags must match ALL of our environments for tag in fileinfo["tags"]: - (intrp_name, intrp_ver) = parse_interpreter(tag.interpreter) + intrp_name, intrp_ver = parse_interpreter(tag.interpreter) if intrp_name not in ("py", "cp"): continue @@ -722,11 +706,7 @@ def mirror(args: argparse.Namespace): m.copy_server() -def main(): # noqa: C901 - """ - Executes the command line interface of Morgan. Use -h for a full list of - flags, options and arguments. - """ +def create_arg_parser() -> argparse.ArgumentParser: def my_url(arg): # url -> url/ without params @@ -802,6 +782,7 @@ def my_url(arg): server.add_arguments(parser) configurator.add_arguments(parser) + Index.add_arguments(parser) parser.add_argument( "command", @@ -816,6 +797,16 @@ def my_url(arg): help="Command to execute", ) + return parser + + +def main(): + """ + Executes the command line interface of Morgan. Use -h for a full list of + flags, options and arguments. + """ + + parser = create_arg_parser() args = parser.parse_args() # These commands do not require a configuration file and therefore should diff --git a/morgan/utils.py b/morgan/utils.py index a2e48cb..86a033f 100644 --- a/morgan/utils.py +++ b/morgan/utils.py @@ -1,13 +1,20 @@ from __future__ import annotations +import base64 +import json +import netrc import os import re +import urllib.parse +import urllib.request from collections import OrderedDict from typing import TYPE_CHECKING, Iterable import dateutil # type: ignore[import-untyped] if TYPE_CHECKING: + import argparse + from packaging.requirements import Requirement @@ -48,6 +55,7 @@ def is_simple_case(self, req: Requirement) -> bool: if not specifier: return True # ruff: noqa: SLF001 + # pylint: disable=protected-access if all(spec.operator in (">", ">=") for spec in specifier._specs): return True return False @@ -151,3 +159,57 @@ def __setitem__(self, key, value): self[key].extend(value) else: super().__setitem__(key, value) + + +class Index: + def __init__(self, args: argparse.Namespace): + self.index_url = args.index_url.rstrip("/") + self.auth_header: tuple[str, str] | None = None + netrc_ = None + if args.netrc_file: + netrc_ = netrc.netrc(args.netrc_file) + elif args.netrc: + netrc_ = netrc.netrc() + if netrc_: + hostname = urllib.parse.urlsplit(self.index_url).hostname + t = netrc_.authenticators(hostname) # login, account, passwd + if t: + # ruff: noqa: UP012 + s = base64.b64encode(f"{t[0]}:{t[2]}".encode("utf-8")).decode("ascii") + self.auth_header = ("Authorization", f"Basic {s}") + + def get(self, pkg_name: str) -> tuple[dict, str]: + # get information about this package from the Simple API in JSON + # format as per PEP 691 + request = urllib.request.Request( # noqa: S310 + f"{self.index_url}/{pkg_name}/", + headers={ + "Accept": "application/vnd.pypi.simple.v1+json", + }, + ) + if self.auth_header: + request.add_header(*self.auth_header) + + with urllib.request.urlopen(request) as response: # noqa: S310 + data = json.load(response) + response_url = str(response.url) + if data: + return (data, response_url) + msg = f"Failed loading metadata: {response}" + raise RuntimeError(msg) + + @staticmethod + def add_arguments(parser: argparse.ArgumentParser): + parser.add_argument( + "--netrc", + dest="netrc", + action="store_true", + help="Must read .netrc for username and password", + ) + + parser.add_argument( + "--netrc-file", + dest="netrc_file", + nargs="?", + help="Specify FILE for netrc", + ) diff --git a/pyproject.toml b/pyproject.toml index 65f0e4b..05e3f12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -93,3 +93,17 @@ ignore = [ [tool.mypy] check_untyped_defs = true follow_untyped_imports = true + +[tool.isort] +profile = "black" + +[tool.pylint] +"messages control".disable = [ + "line-too-long", + "missing-class-docstring", + "missing-function-docstring", + "missing-module-docstring", + "too-few-public-methods", + "too-many-instance-attributes", +] +reports.score = false diff --git a/tests/test_init.py b/tests/test_init.py index d7b65f4..8a75b82 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -1,7 +1,6 @@ # pylint: disable=missing-function-docstring,missing-class-docstring,missing-module-docstring from __future__ import annotations -import argparse import hashlib import os @@ -9,7 +8,13 @@ import packaging.version import pytest -from morgan import PYPI_ADDRESS, Mirrorer, parse_interpreter, parse_requirement, server +from morgan import ( + Mirrorer, + create_arg_parser, + parse_interpreter, + parse_requirement, + server, +) class TestParseInterpreter: @@ -88,13 +93,14 @@ def temp_index_path(self, tmpdir): return tmpdir def test_mirrorer_initialization(self, temp_index_path): - args = argparse.Namespace( - index_path=temp_index_path, - index_url="https://pypi.org/simple/", - config=os.path.join(temp_index_path, "morgan.ini"), - mirror_all_versions=False, - package_type_regex="(whl|zip|tar.gz)", - mirror_all_wheels=False, + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--config", + os.path.join(temp_index_path, "morgan.ini"), + ], ) mirrorer = Mirrorer(args) @@ -108,13 +114,14 @@ def test_mirrorer_initialization(self, temp_index_path): assert not mirrorer.mirror_all_versions def test_server_file_copying(self, temp_index_path): - args = argparse.Namespace( - index_path=temp_index_path, - index_url=PYPI_ADDRESS, - config=os.path.join(temp_index_path, "morgan.ini"), - mirror_all_versions=False, - package_type_regex="(whl|zip|tar.gz)", - mirror_all_wheels=False, + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--config", + os.path.join(temp_index_path, "morgan.ini"), + ], ) mirrorer = Mirrorer(args) @@ -134,13 +141,14 @@ def test_server_file_copying(self, temp_index_path): ) def test_file_hashing(self, temp_index_path): - args = argparse.Namespace( - index_path=temp_index_path, - index_url=PYPI_ADDRESS, - config=os.path.join(temp_index_path, "morgan.ini"), - mirror_all_versions=False, - package_type_regex="(whl|zip|tar.gz)", - mirror_all_wheels=False, + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--config", + os.path.join(temp_index_path, "morgan.ini"), + ], ) mirrorer = Mirrorer(args) @@ -184,14 +192,20 @@ def temp_index_path(self, tmp_path): def make_mirrorer(self, temp_index_path): # Return a function that creates mirrorer instances def _make_mirrorer(mirror_all_versions, mirror_all_wheels=False): - args = argparse.Namespace( - index_path=temp_index_path, - index_url="https://example.com/simple", - config=os.path.join(temp_index_path, "morgan.ini"), - mirror_all_versions=mirror_all_versions, - package_type_regex=r"(whl|zip|tar\.gz)", - mirror_all_wheels=mirror_all_wheels, - ) + args_list = [ + "mirror", + "--index-path", + str(temp_index_path), + "--index-url", + "https://example.com/simple", + "--config", + os.path.join(temp_index_path, "morgan.ini"), + ] + if mirror_all_versions: + args_list.append("--mirror-all-versions") + if mirror_all_wheels: + args_list.append("--mirror-all-wheels") + args = create_arg_parser().parse_args(args_list) return Mirrorer(args) return _make_mirrorer diff --git a/tests/test_init_wheelscore.py b/tests/test_init_wheelscore.py index ebfb3b6..74b99c6 100644 --- a/tests/test_init_wheelscore.py +++ b/tests/test_init_wheelscore.py @@ -3,13 +3,12 @@ # pylint: disable=missing-function-docstring,missing-class-docstring,protected-access # ruff: noqa: SLF001 -import argparse import os from typing import NamedTuple import pytest -from morgan import Mirrorer +from morgan import Mirrorer, create_arg_parser class TestCalculateScoresForWheel: @@ -32,13 +31,17 @@ def temp_index_path(self, tmp_path): @pytest.fixture def mirrorer(self, temp_index_path): - args = argparse.Namespace( - index_path=temp_index_path, - index_url="https://example.com/simple/", - config=os.path.join(temp_index_path, "morgan.ini"), - mirror_all_versions=False, - package_type_regex=r"(whl|zip|tar\.gz)", - mirror_all_wheels=True, + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--index-url", + "https://example.com/simple", + "--config", + os.path.join(temp_index_path, "morgan.ini"), + "--mirror-all-wheels", + ], ) return Mirrorer(args) diff --git a/tests/test_utils.py b/tests/test_utils.py index 87325db..1d627f1 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,6 +1,10 @@ +import base64 +import os + import pytest from packaging.requirements import Requirement +from morgan import Mirrorer, create_arg_parser from morgan.utils import filter_relevant_requirements, is_requirement_relevant @@ -208,3 +212,76 @@ def test_filter_with_empty_environments(self): filtered = filter_relevant_requirements(requirements, environments) assert len(filtered) == 2 + + +def create_auth_header(user: str, passwd: str): + # ruff: noqa: UP012 + s = base64.b64encode(f"{user}:{passwd}".encode("utf-8")).decode("ascii") + return ("Authorization", f"Basic {s}") + + +class TestNetrc: + """Tests for Index's .netrc evaluation""" + + @pytest.fixture + def temp_index_path(self, tmp_path): + # Create minimal config file + config_path = os.path.join(tmp_path, "morgan.ini") + with open(config_path, "w", encoding="utf-8") as f: + f.write( + """ + [env.test_env] + python_version = 3.10 + sys_platform = linux + platform_machine = x86_64 + """, + ) + return tmp_path + + @pytest.fixture + def temp_netrc(self, tmp_path): + # Create minimal config file + fpath = os.path.join(tmp_path, ".netrc") + with open(fpath, "w", encoding="utf-8") as f: + f.write( + """\ +default login xxx password yyy + +machine example.com +login daniel +password qwerty +""", + ) + return fpath + + def test_default(self, temp_index_path, temp_netrc): + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--config", + os.path.join(temp_index_path, "morgan.ini"), + "--netrc-file", + str(temp_netrc), + ], + ) + m = Mirrorer(args) + assert m.index.auth_header == create_auth_header("xxx", "yyy") + + def test_example(self, temp_index_path, temp_netrc): + args = create_arg_parser().parse_args( + [ + "mirror", + "--index-path", + str(temp_index_path), + "--index-url", + "https://example.com/simple", + "--config", + os.path.join(temp_index_path, "morgan.ini"), + "--netrc-file", + str(temp_netrc), + ], + ) + m = Mirrorer(args) + assert m.index.auth_header == create_auth_header("daniel", "qwerty")