Add new osc._private module

osc/_private/__init__.py

@@ -0,0 +1,7 @@
+# This is a private implementation of osc.core that will replace it in the future.
+# The existing osc.core needs to stay for a while to emit deprecation warnings.
+#
+# The cherry-picked imports will be the supported API.
+
+from .package import ApiPackage
+from .request import forward_request

osc/_private/api.py

@@ -0,0 +1,74 @@
+"""
+Functions that communicate with OBS API
+and work with related XML data.
+"""
+
+
+from .. import connection as osc_connection
+from .. import core as osc_core
+
+
+def get(apiurl, path, query=None):
+    """
+    Send a GET request to OBS.
+
+    :param apiurl: OBS apiurl.
+    :type  apiurl: str
+    :param path: URL path segments.
+    :type  path: list(str)
+    :param query: URL query values.
+    :type  query: dict(str, str)
+    :returns: Parsed XML root.
+    :rtype:   xml.etree.ElementTree.Element
+    """
+    assert apiurl
+    assert path
+
+    if not isinstance(path, (list, tuple)):
+        raise TypeError("Argument `path` expects a list of strings")
+
+    url = osc_core.makeurl(apiurl, path, query)
+    with osc_connection.http_GET(url) as f:
+        root = osc_core.ET.parse(f).getroot()
+    return root
+
+
+def find_nodes(root, root_name, node_name):
+    """
+    Find nodes with given `node_name`.
+    Also, verify that the root tag matches the `root_name`.
+
+    :param root: Root node.
+    :type  root: xml.etree.ElementTree.Element
+    :param root_name: Expected (tag) name of the root node.
+    :type  root_name: str
+    :param node_name: Name of the nodes we're looking for.
+    :type  node_name: str
+    :returns: List of nodes that match the given `node_name`.
+    :rtype:   list(xml.etree.ElementTree.Element)
+    """
+    assert root.tag == root_name
+    return root.findall(node_name)
+
+
+def find_node(root, root_name, node_name=None):
+    """
+    Find a single node with given `node_name`.
+    If `node_name` is not specified, the root node is returned.
+    Also, verify that the root tag matches the `root_name`.
+
+    :param root: Root node.
+    :type  root: xml.etree.ElementTree.Element
+    :param root_name: Expected (tag) name of the root node.
+    :type  root_name: str
+    :param node_name: Name of the nodes we're looking for.
+    :type  node_name: str
+    :returns: The node that matches the given `node_name`
+              or the root node if `node_name` is not specified.
+    :rtype:   xml.etree.ElementTree.Element
+    """
+
+    assert root.tag == root_name
+    if node_name:
+        return root.find(node_name)
+    return root

osc/_private/package.py

@@ -0,0 +1,87 @@
+import functools
+
+from .. import core as osc_core
+from . import api
+
+
+@functools.total_ordering
+class PackageBase:
+    def __init__(self, apiurl, project, package):
+        self.apiurl = apiurl
+        self.project = project
+        self.name = package
+
+        self.rev = None
+        self.vrev = None
+        self.srcmd5 = None
+
+        self.linkinfo = None
+        self.files = []
+        directory_node = self._get_directory_node()
+        self._load_from_directory_node(directory_node)
+
+    def __str__(self):
+        return f"{self.project}/{self.name}"
+
+    def __repr__(self):
+        return super().__repr__() + f"({self})"
+
+    def __hash__(self):
+        return hash((self.name, self.project, self.apiurl))
+
+    def __eq__(self, other):
+        return (self.name, self.project, self.apiurl) == (other.name, other.project, other.apiurl)
+
+    def __lt__(self, other):
+        return (self.name, self.project, self.apiurl) < (other.name, other.project, other.apiurl)
+
+    def _get_directory_node(self):
+        raise NotImplementedError
+
+    def _load_from_directory_node(self, directory_node):
+        # attributes
+        self.rev = directory_node.get("rev")
+        self.vrev = directory_node.get("vrev")
+        self.srcmd5 = directory_node.get("srcmd5")
+
+        # files
+        file_nodes = api.find_nodes(directory_node, "directory", "entry")
+        for file_node in file_nodes:
+            self.files.append(osc_core.File.from_xml_node(file_node))
+
+        # linkinfo
+        linkinfo_node = api.find_node(directory_node, "directory", "linkinfo")
+        if linkinfo_node is not None:
+            self.linkinfo = osc_core.Linkinfo()
+            self.linkinfo.read(linkinfo_node)
+            if self.linkinfo.project and not self.linkinfo.package:
+                # if the link points to a package with the same name,
+                # the name is omitted and we want it present for overall sanity
+                self.linkinfo.package = self.name
+
+
+class ApiPackage(PackageBase):
+    def __init__(self, apiurl, project, package, rev=None):
+        # for loading the directory node from the API
+        # the actual revision is loaded from the directory node
+        self.__rev = rev
+        super().__init__(apiurl, project, package)
+
+    def _get_directory_node(self):
+        url_path = ["source", self.project, self.name]
+        url_query = {}
+        if self.__rev:
+            url_query["rev"] = self.__rev
+        return api.get(self.apiurl, url_path, url_query)
+
+
+class LocalPackage(PackageBase):
+    def __init__(self, path):
+        self.dir = path
+        apiurl = osc_core.store_read_apiurl(self.dir)
+        project = osc_core.store_read_project(self.dir)
+        package = osc_core.store_read_package(self.dir)
+        super().__init__(apiurl, project, package)
+
+    def _get_directory_node(self):
+        return osc_core.read_filemeta(self.dir).getroot()

osc/_private/request.py

@@ -0,0 +1,34 @@
+from .. import core as osc_core
+from . import package as osc_package
+
+
+def forward_request(apiurl, request, interactive=True):
+    """
+    Forward the specified `request` to the projects the packages were branched from.
+    """
+    for action in request.get_actions("submit"):
+        package = osc_package.ApiPackage(apiurl, action.tgt_project, action.tgt_package)
+
+        if not package.linkinfo:
+            # not a linked/branched package, can't forward to parent
+            continue
+
+        project = package.linkinfo.project
+        package = package.linkinfo.package
+
+        if interactive:
+            reply = input(f"\nForward request to {project}/{package}? ([y]/n) ")
+            if reply.lower() not in ("y", ""):
+                continue
+
+        msg = f"Forwarded request #{request.reqid} from {request.creator}\n\n{request.description}"
+        new_request_id = osc_core.create_submit_request(
+            apiurl,
+            action.tgt_project,
+            action.tgt_package,
+            project,
+            package,
+            msg,
+        )
+        msg = f"Forwarded request #{request.reqid} from {request.creator} to {project}/{package}: #{new_request_id}"
+        print(msg)

tests/fixtures/packages/osctest/openSUSE:Tools/osc/.osc/_apiurl

@@ -0,0 +1 @@
+http://localhost

tests/fixtures/packages/osctest/openSUSE:Tools/osc/.osc/_files

@@ -0,0 +1,7 @@
+<directory name="osc" rev="373" vrev="339" srcmd5="30ccce6c3a1a4322e79c2935a52af18b">
+  <linkinfo project="openSUSE:Factory" package="osc" srcmd5="1ccbcd1b0b531a37ad75b34b5a1e2e3e" baserev="2c3ae65909d69e0f63113ccfe0e5f3f8" xsrcmd5="6a31b956f9431b0644ad6cf8e845c4e5" lsrcmd5="30ccce6c3a1a4322e79c2935a52af18b"/>
+  <serviceinfo code="succeeded" xsrcmd5="f4a02ee746c0f1d92ecef1d0f04a13b9"/>
+  <entry name="osc-0.182.0.tar.gz" md5="87f040c76f3da86fd7218c972b9df1dc" size="381596" mtime="1662638726"/>
+  <entry name="osc.changes" md5="8262e8219ed149bd15be1891fd38d7d1" size="125625" mtime="1662638727"/>
+  <entry name="osc.spec" md5="6fbfa2d1f451c0e3c5e5927ca5b0a2c5" size="6930" mtime="1662639059"/>
+</directory>

tests/fixtures/packages/osctest/openSUSE:Tools/osc/.osc/_osclib_version

@@ -0,0 +1 @@
+1.0

tests/fixtures/packages/osctest/openSUSE:Tools/osc/.osc/_package

@@ -0,0 +1 @@
+osc

tests/fixtures/packages/osctest/openSUSE:Tools/osc/.osc/_project

@@ -0,0 +1 @@
+openSUSE:Tools

tests/test__private_package.py

@@ -0,0 +1,141 @@
+import os
+import unittest
+
+from osc._private.package import ApiPackage
+from osc._private.package import LocalPackage
+from osc._private.package import PackageBase
+
+from .common import GET
+from .common import OscTestCase
+
+
+FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures", "packages")
+
+
+class PackageBaseMock(PackageBase):
+    def _get_directory_node(self):
+        pass
+
+    def _load_from_directory_node(self, directory_node):
+        pass
+
+
+class TestPackageBase(unittest.TestCase):
+    def setUp(self):
+        self.p1 = PackageBaseMock("http://urlA", "projA", "pkgA")
+
+    def test_str(self):
+        self.assertEqual(str(self.p1), "projA/pkgA")
+
+    def test_repr(self):
+        self.assertTrue(repr(self.p1).endswith("(projA/pkgA)"))
+
+    def test_eq(self):
+        # the same
+        p2 = PackageBaseMock(self.p1.apiurl, self.p1.project, self.p1.name)
+        self.assertEqual(self.p1, p2)
+
+        # package name differs
+        p2 = PackageBaseMock(self.p1.apiurl, self.p1.project, "pkgB")
+        self.assertNotEqual(self.p1, p2)
+
+        # project name differs
+        p2 = PackageBaseMock(self.p1.apiurl, "projB", self.p1.name)
+        self.assertNotEqual(self.p1, p2)
+
+        # baseurl differs
+        p2 = PackageBaseMock("http://urlB", self.p1.project, self.p1.name)
+        self.assertNotEqual(self.p1, p2)
+
+    def test_lt(self):
+        # the same
+        p2 = PackageBaseMock(self.p1.apiurl, self.p1.project, self.p1.name)
+        self.assertFalse(self.p1 < p2)
+
+        # package name differs
+        p2 = PackageBaseMock(self.p1.apiurl, self.p1.project, "pkgB")
+        self.assertTrue(self.p1 < p2)
+
+        # project name differs
+        p2 = PackageBaseMock(self.p1.apiurl, "projB", self.p1.name)
+        self.assertTrue(self.p1 < p2)
+
+        # baseurl differs
+        p2 = PackageBaseMock("http://urlB", self.p1.project, self.p1.name)
+        self.assertTrue(self.p1 < p2)
+
+    def test_hash(self):
+        p2 = PackageBaseMock(self.p1.apiurl, self.p1.project, self.p1.name)
+        self.assertEqual(hash(self.p1), hash(p2))
+
+        packages = set()
+        packages.add(self.p1)
+        # the second instance appears to be there because it has the same hash
+        # it is ok, because we consider such packages equal
+        self.assertIn(p2, packages)
+
+
+class TestLocalPackage(OscTestCase):
+    def _get_fixtures_dir(self):
+        return FIXTURES_DIR
+
+    def test_load(self):
+        path = os.path.join(self.tmpdir, "osctest", "openSUSE:Tools", "osc")
+        p = LocalPackage(path)
+
+        self.assertEqual(p.name, "osc")
+        self.assertEqual(p.project, "openSUSE:Tools")
+        self.assertEqual(p.apiurl, "http://localhost")
+        self.assertEqual(p.rev, "373")
+        self.assertEqual(p.vrev, "339")
+        self.assertEqual(p.srcmd5, "30ccce6c3a1a4322e79c2935a52af18b")
+
+        self.assertEqual(p.linkinfo.project, "openSUSE:Factory")
+        self.assertEqual(p.linkinfo.package, "osc")
+        self.assertEqual(p.linkinfo.srcmd5, "1ccbcd1b0b531a37ad75b34b5a1e2e3e")
+        self.assertEqual(p.linkinfo.baserev, "2c3ae65909d69e0f63113ccfe0e5f3f8")
+        self.assertEqual(p.linkinfo.xsrcmd5, "6a31b956f9431b0644ad6cf8e845c4e5")
+        self.assertEqual(p.linkinfo.lsrcmd5, "30ccce6c3a1a4322e79c2935a52af18b")
+
+        self.assertEqual(len(p.files), 3)
+
+        f = p.files[0]
+        self.assertEqual(f.name, "osc-0.182.0.tar.gz")
+        self.assertEqual(f.md5, "87f040c76f3da86fd7218c972b9df1dc")
+        self.assertEqual(f.size, 381596)
+        self.assertEqual(f.mtime, 1662638726)
+
+
+class TestApiPackage(OscTestCase):
+    def _get_fixtures_dir(self):
+        return FIXTURES_DIR
+
+    @GET("http://localhost/source/openSUSE:Tools/osc", file="osctest/openSUSE:Tools/osc/.osc/_files")
+    def test_load(self):
+        p = ApiPackage("http://localhost", "openSUSE:Tools", "osc")
+
+        self.assertEqual(p.name, "osc")
+        self.assertEqual(p.project, "openSUSE:Tools")
+        self.assertEqual(p.apiurl, "http://localhost")
+        self.assertEqual(p.rev, "373")
+        self.assertEqual(p.vrev, "339")
+        self.assertEqual(p.srcmd5, "30ccce6c3a1a4322e79c2935a52af18b")
+
+        self.assertEqual(p.linkinfo.project, "openSUSE:Factory")
+        self.assertEqual(p.linkinfo.package, "osc")
+        self.assertEqual(p.linkinfo.srcmd5, "1ccbcd1b0b531a37ad75b34b5a1e2e3e")
+        self.assertEqual(p.linkinfo.baserev, "2c3ae65909d69e0f63113ccfe0e5f3f8")
+        self.assertEqual(p.linkinfo.xsrcmd5, "6a31b956f9431b0644ad6cf8e845c4e5")
+        self.assertEqual(p.linkinfo.lsrcmd5, "30ccce6c3a1a4322e79c2935a52af18b")
+
+        self.assertEqual(len(p.files), 3)
+
+        f = p.files[0]
+        self.assertEqual(f.name, "osc-0.182.0.tar.gz")
+        self.assertEqual(f.md5, "87f040c76f3da86fd7218c972b9df1dc")
+        self.assertEqual(f.size, 381596)
+        self.assertEqual(f.mtime, 1662638726)
+
+
+if __name__ == "__main__":
+    unittest.main()