summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/selftest/prservice.py
blob: 789c05f1e552f9af1b7df261fdce3601db2cc3b6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
import unittest
import os
import logging
import re
import shutil
import datetime

import oeqa.utils.ftools as ftools
from oeqa.selftest.base import oeSelfTest
from oeqa.utils.commands import runCmd, bitbake, get_bb_var

class BitbakePrTests(oeSelfTest):

    def get_pr_version(self, package_name):
        pkgdata_dir = get_bb_var('PKGDATA_DIR')
        package_data_file = os.path.join(pkgdata_dir, 'runtime', package_name)
        package_data = ftools.read_file(package_data_file)
        find_pr = re.search("PKGR: r[0-9]+\.([0-9]+)", package_data)
        self.assertTrue(find_pr)
        return int(find_pr.group(1))

    def get_task_stamp(self, package_name, recipe_task):
        stampdata = get_bb_var('STAMP', target=package_name).split('/')
        prefix = stampdata[-1]
        package_stamps_path = "/".join(stampdata[:-1])
        stamps = []
        for stamp in os.listdir(package_stamps_path):
            find_stamp = re.match("%s\.%s\.([a-z0-9]{32})" % (prefix, recipe_task), stamp)
            if find_stamp:
                stamps.append(find_stamp.group(1))
        self.assertFalse(len(stamps) == 0, msg="Cound not find stamp for task %s for recipe %s" % (recipe_task, package_name))
        self.assertFalse(len(stamps) > 1, msg="Found multiple %s stamps for the %s recipe in the %s directory." % (recipe_task, package_name, package_stamps_path))
        return str(stamps[0])

    def increment_package_pr(self, package_name):
        inc_data = "do_package_append() {\nbb.build.exec_func('do_test_prserv', d)\n}\ndo_test_prserv() {\necho \"The current date is: %s\"\n}" % datetime.datetime.now()
        self.write_recipeinc(package_name, inc_data)
        bitbake("-ccleansstate %s" % package_name)
        res = bitbake(package_name, ignore_status=True)
        self.delete_recipeinc(package_name)
        self.assertEqual(res.status, 0, msg=res.output)
        self.assertTrue("NOTE: Started PRServer with DBfile" in res.output, msg=res.output)

    def config_pr_tests(self, package_name, package_type='rpm', pr_socket='localhost:0'):
        config_package_data = 'PACKAGE_CLASSES = "package_%s"' % package_type
        self.write_config(config_package_data)
        config_server_data = 'PRSERV_HOST = "%s"' % pr_socket
        self.append_config(config_server_data)

    def run_test_pr_service(self, package_name, package_type='rpm', track_task='do_package', pr_socket='localhost:0'):
        self.config_pr_tests(package_name, package_type, pr_socket)

        self.increment_package_pr(package_name)
        pr_1 = self.get_pr_version(package_name)
        stamp_1 = self.get_task_stamp(package_name, track_task)

        self.increment_package_pr(package_name)
        pr_2 = self.get_pr_version(package_name)
        stamp_2 = self.get_task_stamp(package_name, track_task)

        bitbake("-ccleansstate %s" % package_name)
        self.assertTrue(pr_2 - pr_1 == 1)
        self.assertTrue(stamp_1 != stamp_2)

    def run_test_pr_export_import(self, package_name, replace_current_db=True):
        self.config_pr_tests(package_name)

        self.increment_package_pr(package_name)
        pr_1 = self.get_pr_version(package_name)

        exported_db_path = os.path.join(self.builddir, 'export.inc')
        export_result = runCmd("bitbake-prserv-tool export %s" % exported_db_path, ignore_status=True)
        self.assertEqual(export_result.status, 0, msg="PR Service database export failed: %s" % export_result.output)

        if replace_current_db:
            current_db_path = os.path.join(get_bb_var('PERSISTENT_DIR'), 'prserv.sqlite3')
            self.assertTrue(os.path.exists(current_db_path), msg="Path to current PR Service database is invalid: %s" % current_db_path)
            os.remove(current_db_path)

        import_result = runCmd("bitbake-prserv-tool import %s" % exported_db_path, ignore_status=True)
        os.remove(exported_db_path)
        self.assertEqual(import_result.status, 0, msg="PR Service database import failed: %s" % import_result.output)

        self.increment_package_pr(package_name)
        pr_2 = self.get_pr_version(package_name)

        bitbake("-ccleansstate %s" % package_name)
        self.assertTrue(pr_2 - pr_1 == 1)


    def test_import_export_replace_db(self):
        self.run_test_pr_export_import('m4')

    def test_import_export_override_db(self):
        self.run_test_pr_export_import('m4', replace_current_db=False)

    def test_pr_service_rpm_arch_dep(self):
        self.run_test_pr_service('m4', 'rpm', 'do_package')

    def test_pr_service_deb_arch_dep(self):
        self.run_test_pr_service('m4', 'deb', 'do_package')

    def test_pr_service_ipk_arch_dep(self):
        self.run_test_pr_service('m4', 'ipk', 'do_package')

    def test_pr_service_rpm_arch_indep(self):
        self.run_test_pr_service('xcursor-transparent-theme', 'rpm', 'do_package')

    def test_pr_service_deb_arch_indep(self):
        self.run_test_pr_service('xcursor-transparent-theme', 'deb', 'do_package')

    def test_pr_service_ipk_arch_indep(self):
        self.run_test_pr_service('xcursor-transparent-theme', 'ipk', 'do_package')