Merge ppa-dev-tools:results-class into ppa-dev-tools:main

[Bryce Harrington]

Adds the Subtest, Trigger, and Results classes.

I've run lint/flake on these via the check-scripts command from ubuntu-helpers.

    $ check-scripts ./ppa/result.py
    $ check-scripts ./tests/test_result.py
    ...

Each module has a corresponding test that can be invoked via pytest-3, e.g:

    $ pytest-3 ./tests/test_result.py
    ...

Each module can also be run as a script, as a cheap smoketest:

    $ python3 -m ppa.result

[Athos Ribeiro (athos-ribeiro)]

Thanks, Bryce. LGTM!

I left a few inline comments, including a question, suggestion on a changed test.

review: Approve

[Bryce Harrington (bryce)]

Answer & re-question for one of your feedback items, inline.

[Athos Ribeiro (athos-ribeiro)]

On Wed, Aug 31, 2022 at 06:40:02PM -0000, Bryce Harrington wrote:
>Answer & re-question for one of your feedback items, inline.
>
>Diff comments:
>
>> diff --git a/tests/test_job.py b/tests/test_job.py
>> index d9a9967..44491c2 100644
>> --- a/tests/test_job.py
>> +++ b/tests/test_job.py
>> @@ -71,9 +74,10 @@ def test_request_url():
>> jobinfo = {
>> 'triggers': ['t/1'],
>> 'ppas': ['ppa:a/b', 'ppa:c/d']
>> - }
>> + }
>> job = Job(0, 'a', 'b', 'c', 'd', jobinfo['triggers'], jobinfo['ppas'])
>> - assert job.request_url == "https://autopkgtest.ubuntu.com/request.cgi?release=c&arch=d&package=b&trigger=t/1"
>> + assert job.request_url.startswith("https://autopkgtest.ubuntu.com/request.cgi")
>> + assert job.request_url.endswith("?release=c&arch=d&package=b&trigger=t/1")
>
>Yeah it was for shortening the length for the lint warning. Normally I'd do the new variables but figured that would make the testing less clear.
>
>My thinking here for two tests is that it now checks a) is the cgi address correctly inserted, and then b) are the args showing up properly. Do you think it'd be better to just check the whole URL?
>

This was just a nitpick to point out that the test changed. To keep the
behavior, i.e, for the commit to be a refactoring only commit, it should
keep checking the whole thing. It should be OK to merge it as is though
:)

>>
>>
>> def test_get_running():
>
>
>--
>https://code.launchpad.net/~bryce/ppa-dev-tools/+git/ppa-dev-tools-1/+merge/429187
>You are reviewing the proposed merge of ppa-dev-tools:results-class into ppa-dev-tools:main.
>

--
Athos Ribeiro

[Bryce Harrington (bryce)]

Thanks Athos, yeah while it's not a pure refactor I think splitting the url check to two tests is a good change. I've gone ahead and landed the branch.

diff --git a/.flake8 b/.flake8
index 7da1f96..d6c1210 100644
--- a/.flake8
+++ b/.flake8
@@ -1,2 +1,3 @@
 [flake8]
 max-line-length = 100
+ignore = E402
diff --git a/AUTHORS.md b/AUTHORS.md
index 6138e91..4aef7e4 100644
--- a/AUTHORS.md
+++ b/AUTHORS.md
@@ -1,2 +1,4 @@
 Bryce Harrington <[e-mail]>
 Athos Ribeiro <[e-mail]>
+Christian Ehrhardt <[e-mail]>
+Andy P. Whitcroft <[e-mail]>
diff --git a/ppa/result.py b/ppa/result.py
new file mode 100755
index 0000000..7a2ca5a
--- /dev/null
+++ b/ppa/result.py
@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Copyright (C) 2022 Authors
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+#
+# Authors:
+#   Bryce Harrington <[e-mail]>
+
+"""The completed data from an autopkgtest run"""
+
+import re
+import urllib.request
+from functools import lru_cache, cached_property
+from typing import List, Iterator
+import gzip
+import time
+
+from .subtest import Subtest
+from .trigger import Trigger
+
+
+class Result:
+    """
+    The completed data from an autopkgtest run Job.  This object
+    provides access to the test run's settings and results.
+    """
+    VALUES = {
+        'PASS': "✅",
+        'FAIL': "❌",
+        'BAD':  "⛔",
+    }
+
+    def __init__(self, url, time, series, arch, source):
+        """Initializes a new Result object.
+
+        :param str url: HTTP path to the test log for this result.
+        :param str time: The execution time of the test run.
+        :param str series: The distro release series codename.
+        :param str arch: The architecture for the result.
+        :param str source:
+        """
+        self.url = url
+        self.time = time
+        self.series = series
+        self.arch = arch
+        self.source = source
+        self.error_message = None
+        self._log = None
+
+    def __repr__(self) -> str:
+        """Machine-parsable unique representation of object.
+
+        :rtype: str
+        :returns: Official string representation of the object.
+        """
+        return (f'{self.__class__.__name__}('
+                f'url={self.url!r})')
+
+    def __str__(self) -> str:
+        """Human-readable summary of the object.
+
+        :rtype: str
+        :returns: Printable summary of the object.
+        """
+        return f"{self.source} on {self.series} for {self.arch} @ {self.timestamp}"
+
+    @property
+    def timestamp(self) -> str:
+        """Formats the result's completion time as a string."""
+        return time.strftime("%d.%m.%y %H:%M:%S", self.time)
+
+    @cached_property
+    def log(self) -> str:
+        """Returns log contents for results, downloading if necessary.
+
+        Retrieves the log via the result url, handles decompression, and
+        caches the results internally, so that subsequent calls don't
+        re-download the data.
+
+        On error, returns None and stores the error message in
+        the Result.error_message property.
+
+        :rtype: str
+        :returns: Full text of the log file, or None on error.
+        """
+        request = urllib.request.Request(self.url)
+        request.add_header('Cache-Control', 'max-age=0')
+        try:
+            response = urllib.request.urlopen(request)
+        except urllib.error.HTTPError as e:
+            self.error_message = f"Failed to Download Test Log ⚪: {e}"
+            return None
+
+        result_gzip = response.read()
+        try:
+            return gzip.decompress(result_gzip).decode("utf-8",
+                                                       errors="replace")
+        except UnicodeDecodeError:
+            self.error_message = "Broken Test Log ⚪"
+            return None
+
+    # TODO: Merge triggers and get_triggers()
+    @cached_property
+    def triggers(self) -> List[str]:
+        """Returns package/version parameters used to generate this Result.
+
+        This returns the set of triggers used to create the Result, as
+        recorded in the test log.  Each trigger is a package/version
+        pair corresponding to source packages to use from the proposed
+        archive (instead of from the release archive).
+
+        :rtype: list[str]
+        :returns: List of package/version triggers.
+        """
+        regex_triggers = re.compile(r'--env=ADT_TEST_TRIGGERS=(.*?) -- ')
+        header_split = self.log.split(": @@@@@@@@@@@@@@@@@@@@", 1)
+        if len(header_split) < 2:
+            return []
+        m = re.search(regex_triggers, header_split[0])
+        if not m:
+            return []
+
+        return m.group(1).strip("'").split()
+
+    @lru_cache
+    def get_triggers(self, name=None) -> Iterator[Trigger]:
+        """Provides list of Triggers that were used to create this Result.
+
+        This returns the set of Triggers used to create the Result, as
+        recorded in the test log.  Each trigger identifies a
+        package/version pair corresponding to source packages to use
+        from the proposed archive (instead of from the release archive).
+
+        :param str name: If defined, only return triggers starting with this name.
+        :rtype: Iterator[Trigger]
+        :returns: Triggers used to generate this Result, if any, or an empty list
+        """
+        if not self.triggers:
+            return []
+
+        for t in self.triggers:
+            package, version = t.split('/', 1)
+            yield Trigger(package, version, arch=self.arch, series=self.series)
+
+
+    @lru_cache
+    def get_subtests(self, name=None) -> Iterator[Subtest]:
+        """Provides list of Subtests that were run for this Result.
+
+        :param str name: Only display subtests starting with this.
+        :rtype: Iterator[Subtest]
+        :returns: Subtests completed for this Result, or empty list.
+        """
+        result_split = self.log.split("@@@@@@@@@@@@@@@@@@@@ summary", 1)
+        if len(result_split) < 2:
+            return []
+
+        result_sum = result_split[1]
+        for line in re.findall("(.*PASS|.*SKIP|.*FAIL|.*BAD)", result_sum):
+            if name and not line.startswith(name):
+                continue
+            yield Subtest(line)
+
+    @cached_property
+    def status(self) -> str:
+        """Returns overall status of all subtests
+
+        If the triggered run completed successfully, then the status will
+        be either FAIL if any of the subtests failed, or PASS otherwise.
+
+        If the run did not complete successfully, then a 'BAD' status
+        will be returned, and the reason can be examined via the
+        Result.error_message property.
+
+        :rtype: str
+        :returns: 'PASS', 'FAIL', or 'BAD'
+        """
+        if self.error_message:
+            return 'BAD'
+
+        for subtest in self.get_subtests():
+            if subtest.status == 'FAIL':
+                return 'FAIL'
+        return 'PASS'
+
+    @cached_property
+    def status_icon(self) -> str:
+        """Unicode symbol corresponding to test's overall status.
+
+        :rtype: str
+        :returns: Unicode symbol
+        """
+        return Result.VALUES[self.status]
+
+
+def get_results(response, base_url, arches=None, sources=None) -> Iterator[Result]:
+    """Returns iterator of Results from the base URL for given criteria
+
+    Retrieves the autopkgtest results limited to the given architectures
+    and source packages.  If unspecified, returns all results.
+
+    :param str base_url: URL for the autopkgtest results.
+    :param list[str] arches: Architectures to include in results.
+    :param list[str] sources: Source packages to include in results.
+    :rtype: Iterator[Result]
+    :returns: Iterable results, if any, or an empty list on error
+    """
+    for line in response.read().split(b'\n'):
+        if line == b'' or not line.endswith(b"log.gz"):
+            continue
+        result = line.decode("utf-8")
+        series, arch, _, source, timestamp = result.split('/')[:5]
+        if (arches and (arch not in arches)):
+            continue
+        if (sources and (source not in sources)):
+            continue
+        yield Result(
+            url=base_url + result,
+            time=time.strptime(timestamp[:-7], "%Y%m%d_%H%M%S"),
+            series=series,
+            arch=arch,
+            source=source)
+
+
+if __name__ == "__main__":
+    import os
+
+    from ppa.io import open_url
+    from ppa.constants import ARCHES_AUTOPKGTEST, URL_AUTOPKGTEST
+
+    print("### Result class smoke test ###")
+
+    timestamp = time.strptime('20030201_040506', "%Y%m%d_%H%M%S")
+    result_1 = Result('url-here', timestamp, 'kinetic', 'amd64', 'my-package')
+    print(repr(result_1))
+    print(result_1)
+    print()
+
+    data_dir = os.path.realpath(os.path.join(os.path.dirname(__file__), "../tests/data"))
+    url = f"file://{data_dir}/results-six-s390x.log.gz"
+    result_2 = Result(url, timestamp, 'kinetic', 'amd64', 'my-package')
+    print("Log Head:")
+    print("\n".join(result_2.log.splitlines()[0:4]))
+    print()
+
+    # TODO: Implement something that dumps the passing tests for given package from -proposed
+    # TODO: Filter to items with only Pass, Not a regression, or No test results
+
+    print("Loading live excuses data")
+    base_results_fmt = f"{URL_AUTOPKGTEST}/results/autopkgtest-%s-%s-%s/"
+    base_results_url = base_results_fmt % ('kinetic', 'bryce', 'nginx-merge-v1.22.0-1')
+    url = f"{base_results_url}?format=plain"
+    response = open_url(url)
+
+    for result in get_results(response, base_results_url, arches=ARCHES_AUTOPKGTEST):
+        print(f"* {result}")
+        print(f"  - Triggers: " + ', '.join([str(r) for r in result.get_triggers()]))
+
+        for subtest in result.get_subtests():
+            print(f"  - {subtest}")
+
+        print()
diff --git a/ppa/subtest.py b/ppa/subtest.py
new file mode 100755
index 0000000..c63dda6
--- /dev/null
+++ b/ppa/subtest.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Copyright (C) 2022 Authors
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+#
+# Authors:
+#   Bryce Harrington <[e-mail]>
+
+"""An individual DEP8 test run"""
+
+from functools import cached_property
+from typing import List, Iterator
+
+
+class Subtest:
+    """
+    A triggered autopkgtest can invoke multiple DEP8 tests, such as running
+    checks on dependencies, the software's testsuite, and integration tests.
+    Each of these is considered a "Subtest"
+    """
+    VALUES = {
+        'PASS': "🟩",
+        'SKIP': "🟧",
+        'FAIL': "🟥",
+        'BAD':  "⛔",
+        'UNKNOWN': "⚪"
+        }
+
+    def __init__(self, line):
+        """Initializes a new Subtext object.
+
+        :param str line: The subtest result summary from a test log.
+        """
+        if not line:
+            raise ValueError("undefined line.")
+
+        self._line = line
+
+    def __repr__(self) -> str:
+        """Machine-parsable unique representation of object.
+
+        :rtype: str
+        :returns: Official string representation of the object.
+        """
+        return (f'{self.__class__.__name__}('
+                f'line={self._line!r})')
+
+    def __str__(self) -> str:
+        """Human-readable summary of the object.
+
+        :rtype: str
+        :returns: Printable summary of the object.
+        """
+        return f"{self.status_icon} {self.desc}: {self.status}"
+
+    @cached_property
+    def desc(self) -> str:
+        """The descriptive text for the given subtest.
+
+        :rtype: str
+        :returns: Descriptive text.
+        """
+        d = self._line.rsplit(':', 1)
+        if not d:
+            return ''
+        return d[0]
+
+    @cached_property
+    def status(self) -> str:
+        """The success or failure of the given subtest.
+
+        :rtype: str
+        :returns: Status term in capitalized letters (PASS, FAIL, etc.)
+        """
+        for k in Subtest.VALUES:
+            if self._line.endswith(k):
+                return k
+        return 'UNKNOWN'
+
+    @cached_property
+    def status_icon(self) -> str:
+        """Unicode symbol corresponding to subtest's status.
+
+        :rtype: str
+        :returns: Single unicode character matching the status.
+        """
+        return Subtest.VALUES[self.status]
+
+
+if __name__ == "__main__":
+    print("### Subtest class smoke test ###")
+    print()
+
+    print("Valid cases")
+    print("-----------")
+    print(Subtest('subtest a: UNKNOWN'))
+    print(Subtest('subtest b: PASS'))
+    print(Subtest('subtest c: FAIL'))
+    print(Subtest('subtest d: SKIP'))
+    print(Subtest('subtest e: BAD'))
+    print()
+
+    print("Invalid cases")
+    print("-------------")
+    print(Subtest('invalid subtest: invalid'))
+    print(Subtest('bAd subtest: bAd'))
diff --git a/ppa/trigger.py b/ppa/trigger.py
new file mode 100755
index 0000000..79276da
--- /dev/null
+++ b/ppa/trigger.py
@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Copyright (C) 2022 Authors
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+#
+# Authors:
+#   Bryce Harrington <[e-mail]>
+
+"""A directive to run a DEP8 test against a source package"""
+
+from functools import cached_property
+
+from .constants import URL_AUTOPKGTEST
+
+
+class Trigger:
+    """
+    A trigger indicates a source package that should be installed from a
+    given series (generally to a specific version) when running
+    autopkgtests for a Job.
+
+    A Job can have multiple Triggers, each against a different source
+    package and/or architectures, but all such Triggers must be against
+    the same series as the Job itself.
+    """
+    def __init__(self, package, version, arch, series):
+        """Initializes a new Trigger for a given package and version.
+
+        :param str package: The source package name.
+        :param str version: The version of the source package to install.
+        :param str arch: The architecture for the trigger.
+        :param str series: The distro release series codename.
+        """
+        self.package = package
+        self.version = version
+        self.arch = arch
+        self.series = series
+
+    def __repr__(self) -> str:
+        """Machine-parsable unique representation of object.
+
+        :rtype: str
+        :returns: Official string representation of the object.
+        """
+        return (f'{self.__class__.__name__}('
+                f'package={self.package!r}, version={self.version!r}, '
+                f'arch={self.arch!r}, series={self.series!r})')
+
+    def __str__(self) -> str:
+        """Human-readable summary of the object.
+
+        :rtype: str
+        :returns: Printable summary of the object.
+        """
+        return f"{self.package}/{self.version}"
+
+    @cached_property
+    def autopkgtest_url(self) -> str:
+        """Renders the trigger as a URL to the job history.
+
+        :rtype: str
+        :returns: tbd
+        """
+        if self.package.startswith('lib'):
+            prefix = self.package[0:4]
+        else:
+            prefix = self.package[0]
+        pkg_str = f"{prefix}/{self.package}"
+        return f"{URL_AUTOPKGTEST}/packages/{pkg_str}/{self.series}/{self.arch}"
+
+
+if __name__ == "__main__":
+    print("### Trigger class smoke test ###")
+
+    trigger = Trigger('my-package', '1.2.3', 'amd64', 'kinetic')
+    print(trigger)
+    print(trigger.autopkgtest_url)
diff --git a/tests/data/results-six-s390x.log.gz b/tests/data/results-six-s390x.log.gz
new file mode 100644
index 0000000..70d1a80
Binary files /dev/null and b/tests/data/results-six-s390x.log.gz differ
diff --git a/tests/test_io.py b/tests/test_io.py
index fce6611..57f8737 100644
--- a/tests/test_io.py
+++ b/tests/test_io.py
@@ -13,10 +13,9 @@
 import os
 import sys
 import urllib
-import pytest
 
 sys.path.insert(0, os.path.realpath(
-        os.path.join(os.path.dirname(__file__), "..")))
+    os.path.join(os.path.dirname(__file__), "..")))
 
 from ppa.io import open_url
 
@@ -28,5 +27,5 @@ def test_open_url(tmp_path):
 
     request = open_url(f"file://{f}")
     assert request
-    assert type(request) == urllib.response.addinfourl
+    assert isinstance(request, urllib.response.addinfourl)
     assert request.read().decode() == 'abcde'
diff --git a/tests/test_job.py b/tests/test_job.py
index d9a9967..44491c2 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -14,16 +14,18 @@ import os
 import sys
 
 sys.path.insert(0, os.path.realpath(
-        os.path.join(os.path.dirname(__file__), "..")))
+    os.path.join(os.path.dirname(__file__), "..")))
 
 from ppa.job import Job, get_running, get_waiting
 
 
 class ResponseMock:
+    """Synthetic response object"""
     def __init__(self, text):
         self._text = text.encode('utf-8')
 
     def read(self):
+        """Simply returns the exact text provided in initializer."""
         return self._text
 
 
@@ -34,6 +36,7 @@ def test_object():
 
 
 def test_init():
+    """Checks the initialization of a Job object."""
     job = Job(0, 'a', 'b', 'c', 'd')
     assert job.number == 0
     assert job.submit_time == 'a'
@@ -71,9 +74,10 @@ def test_request_url():
     jobinfo = {
         'triggers': ['t/1'],
         'ppas': ['ppa:a/b', 'ppa:c/d']
-        }
+    }
     job = Job(0, 'a', 'b', 'c', 'd', jobinfo['triggers'], jobinfo['ppas'])
-    assert job.request_url == "https://autopkgtest.ubuntu.com/request.cgi?release=c&arch=d&package=b&trigger=t/1"
+    assert job.request_url.startswith("https://autopkgtest.ubuntu.com/request.cgi")
+    assert job.request_url.endswith("?release=c&arch=d&package=b&trigger=t/1")
 
 
 def test_get_running():
diff --git a/tests/test_result.py b/tests/test_result.py
new file mode 100644
index 0000000..836bbd1
--- /dev/null
+++ b/tests/test_result.py
@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Author:  Bryce Harrington <[e-mail]>
+#
+# Copyright (C) 2022 Bryce W. Harrington
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+
+"""Results class tests"""
+
+import os
+import sys
+import time
+
+import gzip
+
+sys.path.insert(0, os.path.realpath(
+    os.path.join(os.path.dirname(__file__), "..")))
+
+from ppa.result import Result
+
+
+def test_object():
+    """Checks that Result objects can be instantiated."""
+    result = Result('url', None, 'b', 'c', 'd')
+    assert result
+
+
+def test_repr():
+    """Checks Result object representation."""
+    result = Result('url', None, 'b', 'c', 'd')
+    # TODO: Should this include the full set of args?
+    assert repr(result) == "Result(url='url')"
+
+
+def test_str():
+    """Checks Result object textual presentation."""
+    timestamp = time.strptime('20030201_040506', "%Y%m%d_%H%M%S")
+    result = Result('url', timestamp, 'b', 'c', 'd')
+    assert f"{result}" == 'd on b for c @ 01.02.03 04:05:06'
+
+
+def test_timestamp():
+    """Checks Result object formats the result's time correctly."""
+    timestamp = time.strptime('20030201_040506', "%Y%m%d_%H%M%S")
+    result = Result('url', timestamp, 'b', 'c', 'd')
+    assert f"{result.timestamp}" == '01.02.03 04:05:06'
+
+
+def test_log(tmp_path):
+    """Checks that the log content of a Result is available."""
+    f = tmp_path / "result.log.gz"
+    compressed_text = gzip.compress(bytes('abcde', 'utf-8'))
+    f.write_bytes(compressed_text)
+
+    result = Result(f"file://{f}", None, None, None, None)
+    assert result.log == "abcde"
+
+
+def test_triggers():
+    """Checks that autopkgtest triggers can be extracted from test result logs."""
+    data_dir = "/home/bryce/src/PpaDevTools/ppa-dev-tools-result-class/tests/data"
+    result = Result(f"file://{data_dir}/results-six-s390x.log.gz", None, None, None, None)
+    assert result.triggers == ['pygobject/3.42.2-2', 'six/1.16.0-4']
diff --git a/tests/test_subtest.py b/tests/test_subtest.py
new file mode 100644
index 0000000..6ca35af
--- /dev/null
+++ b/tests/test_subtest.py
@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Author:  Bryce Harrington <[e-mail]>
+#
+# Copyright (C) 2022 Bryce W. Harrington
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+
+"""Subtest class tests"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.realpath(
+    os.path.join(os.path.dirname(__file__), "..")))
+
+from ppa.subtest import Subtest
+
+
+def test_object():
+    """Checks that Subtest objects can be instantiated."""
+    subtest = Subtest('a: UNKNOWN')
+    # TODO: If no ':' in description, should throw exception?
+    # TODO: Or add a function that parses lines and returns subtests?
+    assert subtest
+
+
+def test_repr():
+    """Checks Subtest object representation."""
+    subtest = Subtest('a: PASS')
+    assert repr(subtest) == "Subtest(line='a: PASS')"
+
+
+def test_str():
+    """Checks Subtest object textual presentation."""
+    subtest = Subtest('a: PASS')
+    assert f"{subtest}" == '🟩 a: PASS'
+
+
+def test_desc():
+    """Checks Subtest description is parsed correctly."""
+    subtest = Subtest('a: PASS')
+    assert subtest.desc == 'a'
+
+    subtest = Subtest('a:b : c:d: PASS')
+    assert subtest.desc == 'a:b : c:d'
+
+
[e-mail]('line, status', [
+    ('a: PASS', 'PASS'),
+    ('b: SKIP', 'SKIP'),
+    ('c: FAIL', 'FAIL'),
+    ('d: BAD', 'BAD'),
+    ('e: UNKNOWN', 'UNKNOWN'),
+    ('f: invalid', 'UNKNOWN'),
+    ('f: bAd', 'UNKNOWN'),
+])
+def test_status(line, status):
+    """Checks Subtest status is parsed correctly."""
+    subtest = Subtest(line)
+    assert subtest.status == status
+
+    subtest = Subtest('a: PASS')
+
+
[e-mail]('line, icon', [
+    ('a: PASS', "🟩"),
+    ('b: SKIP', "🟧"),
+    ('c: FAIL', "🟥"),
+    ('d: BAD', "⛔"),
+    ('e: UNKNOWN', "⚪"),
+    ('f: invalid', "⚪"),
+    ('f: bAd', "⚪"),
+])
+def test_status_icon(line, icon):
+    """Checks Subtest provides correct icon for status.
+
+    :param str line: Subtest status line to be parsed.
+    :param str icon: Resulting status icon that should be returned.
+    """
+    subtest = Subtest(line)
+    assert subtest.status_icon == icon
diff --git a/tests/test_trigger.py b/tests/test_trigger.py
new file mode 100644
index 0000000..c6b6a6b
--- /dev/null
+++ b/tests/test_trigger.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
+
+# Author:  Bryce Harrington <[e-mail]>
+#
+# Copyright (C) 2022 Bryce W. Harrington
+#
+# Released under GNU GPLv2 or later, read the file 'LICENSE.GPLv2+' for
+# more information.
+
+"""Trigger class tests"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.realpath(
+    os.path.join(os.path.dirname(__file__), "..")))
+
+from ppa.trigger import Trigger
+
+
+def test_object():
+    """Checks that Trigger objects can be instantiated."""
+    trigger = Trigger('a', 'b', 'c', 'd')
+    assert trigger
+
+
+def test_repr():
+    """Checks Trigger object representation."""
+    trigger = Trigger('a', 'b', 'c', 'd')
+    assert repr(trigger) == "Trigger(package='a', version='b', arch='c', series='d')"
+
+
+def test_str():
+    """Checks Trigger object textual presentation."""
+    trigger = Trigger('a', 'b', 'c', 'd')
+    assert f"{trigger}" == 'a/b'
+
+    trigger = Trigger('dovecot', '1:2.3.19.1+dfsg1-2ubuntu2', 'i386', 'kinetic')
+    assert f"{trigger}" == 'dovecot/1:2.3.19.1+dfsg1-2ubuntu2'
+
+
+def test_autopkgtest_url():
+    """Checks that Trigger objects generate valid autopkgtest urls"""
+    trigger = Trigger('a', 'b', 'c', 'd')
+    expected = "https://autopkgtest.ubuntu.com/packages/a/a/d/c"
+    assert trigger.autopkgtest_url == expected
+
+    trigger = Trigger('apache2', '2.4', 'amd64', 'kinetic')
+    expected = "https://autopkgtest.ubuntu.com/packages/a/apache2/kinetic/amd64"
+    assert trigger.autopkgtest_url == expected
+
+    trigger = Trigger('libwebsockets', '4.1.6-3', 'armhf', 'jammy')
+    expected = "https://autopkgtest.ubuntu.com/packages/libw/libwebsockets/jammy/armhf"
+    assert trigger.autopkgtest_url == expected