Add RepoFlags class and api.get_repo_flags method

4 files changed, 239 insertions, 0 deletions

diff --git a/obsapi/core.py b/obsapi/core.py
index f09ce3b..8845016 100644
--- a/obsapi/core.py
+++ b/obsapi/core.py
@@ -7,6 +7,7 @@ from datetime import datetime
 from obsapi.sourceapi import ObsSourceApi
 from obsapi.buildapi import ObsBuildApi
 from obsapi.formatter import Formatter
+from obsapi.repoflags import RepoFlags
 
 
 LSItem = namedtuple('LSItem', 'name md5 size mtime')
@@ -298,6 +299,9 @@ class ObsApi(object):
 
         return vendor
 
+    def get_repo_flags(self, flag_type, prj, pkg=None):
+        return RepoFlags(flag_type, self.get_meta(prj, pkg))
+
     def locked(self, prj, pkg=None):
         meta = self.__xml2etree(self.get_meta(prj, pkg))
         if meta.find('lock') is not None:
diff --git a/obsapi/repoflags.py b/obsapi/repoflags.py
new file mode 100644
index 0000000..062f0e3
--- /dev/null
+++ b/obsapi/repoflags.py
@@ -0,0 +1,99 @@
+from lxml import etree
+from collections import namedtuple
+
+
+RepoFlag = namedtuple('RepoFlag', 'setting repo arch')
+valid_flag_types = ('build', 'publish', 'useforbuild', 'debuginfo')
+
+
+class RepoFlags():
+
+    def __init__(self, flag_type=None, xml=None):
+
+        self.flag_type = flag_type
+        self.root = None
+
+        if isinstance(xml, (str,)):
+            try:
+                self.import_xml(xml)
+            except Exception:
+                raise Exception('Failed to import xml')
+        elif isinstance(xml, etree._Element):
+            self.root = xml
+
+        if self.flag_type is None and self.root is None:
+            self.flag_type = 'build'
+
+        self.__verify()
+
+    def import_xml(self, xml):
+        try:
+            self.root = etree.fromstring(xml)
+        except Exception as e:
+            raise e
+
+    def __verify(self):
+        """ Make sure flag_type and xml data match.
+        """
+
+        # if self.flag_type set, ensure it matches the tag of the root xml
+        # element. If not matching, try to find the matching element within
+        # the xml tree and set root to that. If not found, set root to a new
+        # empty tree based on the flag_type
+        if self.flag_type is not None:
+            if self.flag_type not in valid_flag_types:
+                raise ValueError('Invalid repo flag type %s' % self.flag_type)
+            if self.root is None:
+                self.root = etree.Element(self.flag_type)
+                return
+            if self.flag_type == self.root.tag:
+                return
+
+            element = self.root.find(self.flag_type)
+            if element is not None:
+                self.root = element
+                return
+
+        # If flag_type is not set, set it to the root element tag
+        flag_type = self.root.tag
+        if flag_type in valid_flag_types:
+            return
+
+        # if root element is not a valid repo flag type, search for the
+        # first element that is a valid flag type and use it.
+        for flag_type in valid_flag_types:
+            element = self.root.find(flag_type)
+            if element is not None:
+                self.flag_type = flag_type
+                self.root = element
+                return
+
+        self.flag_type = 'build'
+        self.root = etree.Element(self.flag_type)
+
+    @property
+    def flags(self):
+        for element in self.root:
+            if element.tag not in ['enable', 'disable']:
+                continue
+            setting = element.tag
+            repo = element.get('repository', None)
+            arch = element.get('arch', None)
+            yield(RepoFlag(setting=setting, repo=repo, arch=arch))
+
+    def set(self, setting, repo=None, arch=None):
+        if setting not in ['enable', 'disable']:
+            raise ValueError('Invalid Flag Setting %s' % setting)
+
+        element = etree.Element(setting)
+        if repo:
+            element.set('repository', repo)
+        if arch:
+            element.set('arch', arch)
+        self.root.append(element)
+
+    def enable(self, repo=None, arch=None):
+        self.set('enable', repo, arch)
+
+    def disable(self, repo=None, arch=None):
+        self.set('disable', repo, arch)
diff --git a/test/test_api.py b/test/test_api.py
index 1830a3d..9f57364 100644
--- a/test/test_api.py
+++ b/test/test_api.py
@@ -101,6 +101,21 @@ def test_get_binary_fileinfo():
     bfinfo = api.get_binary_fileinfo(prj, pkg, repo, arch, binary)
 
 
+def check_get_repo_flags(flag_type, prj, pkg, expected):
+    repo_flags = api.get_repo_flags(flag_type, prj, pkg)
+    assert repo_flags.flag_type == flag_type
+    assert len(list(repo_flags.flags)) == expected
+
+
+def test_get_repo_build_flags():
+    xml = api.get_meta(prj)
+    check_get_repo_flags('build', prj, None, 0)
+    xml = api.get_meta(prj, pkg)
+    meta = etree.fromstring(xml)
+    build = meta.find('build')
+    check_get_repo_flags('build', prj, pkg, len(build))
+
+
 def test_lock_project():
     api.lock(prj)
     xml = api.get_meta(prj)
diff --git a/test/test_repo_flags.py b/test/test_repo_flags.py
new file mode 100644
index 0000000..08a39d6
--- /dev/null
+++ b/test/test_repo_flags.py
@@ -0,0 +1,121 @@
+import pytest
+from obsapi.repoflags import RepoFlags
+
+xml1 = """<project name="home:sbahling:obsapi:test">
+  <title>Test project for obsapi unit tests</title>
+  <description/>
+  <person userid="sbahling" role="maintainer"/>
+  <repository name="SLE_15_SP1">
+    <path project="SUSE:SLE-15-SP1:GA" repository="standard"/>
+    <arch>x86_64</arch>
+  </repository>
+  <repository name="SLE_12_SP3">
+    <path project="SUSE:SLE-12-SP3:GA" repository="standard"/>
+    <arch>x86_64</arch>
+  </repository>
+</project>"""
+
+
+xml2 = """<package name="suse-hello-1.0" project="home:sbahling:obsapi:test">
+  <title>Example from Kernel Module Packages Manual</title>
+  <description/>
+  <build>
+    <enable arch="x86_64" repository="SLE_12_SP3"/>
+    <disable/>
+  </build>
+  <debuginfo>
+    <enable/>
+  </debuginfo>
+  <useforbuild>
+    <disable arch="x86_64" repository="SLE_15_SP1"/>
+  </useforbuild>
+</package>"""
+
+flag_types = ['build', 'publish', 'useforbuild', 'debuginfo']
+
+
+def test_repo_flags_default_type():
+    repo_flags = RepoFlags()
+    assert repo_flags.flag_type == 'build'
+
+
+@pytest.mark.parametrize('flag_type', flag_types)
+def test_repo_flags_flag_type(flag_type):
+    repo_flags = RepoFlags(flag_type)
+    assert repo_flags.flag_type == flag_type
+    assert repo_flags.root.tag == flag_type
+
+
+def test_repo_flags_invalid_flag_type():
+    with pytest.raises(ValueError):
+        RepoFlags('invalid')
+
+
+def test_repo_flags_pass_xml_without_flags():
+    repo_flags = RepoFlags(xml=xml1)
+    assert repo_flags.flag_type == 'build'
+
+
+def test_repo_flags_pass_xml_with_flags():
+    repo_flags = RepoFlags(xml=xml2)
+    assert repo_flags.flag_type in flag_types
+
+
+def test_repo_flags_xml_build_flags():
+    repo_flags = RepoFlags('build', xml=xml2)
+    assert repo_flags.flag_type == 'build'
+    for flag in repo_flags.flags:
+        if flag.setting == 'enable':
+            assert flag.repo == 'SLE_12_SP3'
+            assert flag.arch == 'x86_64'
+        if flag.setting == 'disable':
+            assert flag.repo is None
+            assert flag.arch is None
+
+
+def test_repo_flags_xml_debuginfo_flags():
+    repo_flags = RepoFlags('debuginfo', xml=xml2)
+    assert repo_flags.flag_type == 'debuginfo'
+    for flag in repo_flags.flags:
+        if flag.setting == 'enable':
+            assert flag.repo is None
+            assert flag.arch is None
+
+
+def test_repo_flags_xml_new_flags():
+    repo_flags = RepoFlags('publish', xml=xml2)
+    assert repo_flags.flag_type == 'publish'
+    assert len(list(repo_flags.flags)) == 0
+
+
+params = [('useforbuild', 'SLE_15_SP1', 'x86_64'),
+          ('publish', None, None),
+          ('debuginfo', None, 'aarch64'),
+          ('build', 'SLE_12_SP3', None)
+          ]
+
+
+@pytest.mark.parametrize('flag_type, repo, arch', params)
+def test_repo_flags_enable(flag_type, repo, arch):
+    repoflags = RepoFlags(flag_type)
+    repoflags.enable(repo=repo, arch=arch)
+    for flag in repoflags.flags:
+        assert flag.setting == 'enable'
+        assert flag.repo == repo
+        assert flag.arch == arch
+
+
+@pytest.mark.parametrize('flag_type, repo, arch', params)
+def test_repo_flags_disable(flag_type, repo, arch):
+    repoflags = RepoFlags(flag_type)
+    repoflags.disable(repo=repo, arch=arch)
+    for flag in repoflags.flags:
+        assert flag.setting == 'disable'
+        assert flag.repo == repo
+        assert flag.arch == arch
+
+
+def test_repo_flags_set_invalid():
+    repoflags = RepoFlags()
+    with pytest.raises(ValueError):
+        repoflags.set('invalid')