xref: /openbmc/openbmc/poky/bitbake/lib/prserv/client.py (revision edff49234e31f23dc79f823473c9e286a21596c1) 
1#
2# Copyright BitBake Contributors
3#
4# SPDX-License-Identifier: GPL-2.0-only
5#
6
7import logging
8import bb.asyncrpc
9from . import create_async_client
10
11logger = logging.getLogger("BitBake.PRserv")
12
13class PRAsyncClient(bb.asyncrpc.AsyncClient):
14    def __init__(self):
15        super().__init__("PRSERVICE", "1.0", logger)
16
17    async def getPR(self, version, pkgarch, checksum, history=False):
18        response = await self.invoke(
19            {"get-pr": {"version": version, "pkgarch": pkgarch, "checksum": checksum, "history": history}}
20        )
21        if response:
22            return response["value"]
23
24    async def test_pr(self, version, pkgarch, checksum, history=False):
25        response = await self.invoke(
26            {"test-pr": {"version": version, "pkgarch": pkgarch, "checksum": checksum, "history": history}}
27        )
28        if response:
29            return response["value"]
30
31    async def test_package(self, version, pkgarch):
32        response = await self.invoke(
33            {"test-package": {"version": version, "pkgarch": pkgarch}}
34        )
35        if response:
36            return response["value"]
37
38    async def max_package_pr(self, version, pkgarch):
39        response = await self.invoke(
40            {"max-package-pr": {"version": version, "pkgarch": pkgarch}}
41        )
42        if response:
43            return response["value"]
44
45    async def importone(self, version, pkgarch, checksum, value):
46        response = await self.invoke(
47            {"import-one": {"version": version, "pkgarch": pkgarch, "checksum": checksum, "value": value}}
48        )
49        if response:
50            return response["value"]
51
52    async def export(self, version, pkgarch, checksum, colinfo, history=False):
53        response = await self.invoke(
54            {"export": {"version": version, "pkgarch": pkgarch, "checksum": checksum, "colinfo": colinfo, "history": history}}
55        )
56        if response:
57            return (response["metainfo"], response["datainfo"])
58
59    async def is_readonly(self):
60        response = await self.invoke(
61            {"is-readonly": {}}
62        )
63        if response:
64            return response["readonly"]
65
66class PRClient(bb.asyncrpc.Client):
67    def __init__(self):
68        super().__init__()
69        self._add_methods("getPR", "test_pr", "test_package", "max_package_pr", "importone", "export", "is_readonly")
70
71    def _get_async_client(self):
72        return PRAsyncClient()
73