summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/selftest/prservice.py
diff options
context:
space:
mode:
Diffstat (limited to 'meta/lib/oeqa/selftest/prservice.py')
-rw-r--r--meta/lib/oeqa/selftest/prservice.py113
1 files changed, 113 insertions, 0 deletions
diff --git a/meta/lib/oeqa/selftest/prservice.py b/meta/lib/oeqa/selftest/prservice.py
new file mode 100644
index 0000000..789c05f
--- /dev/null
+++ b/meta/lib/oeqa/selftest/prservice.py
@@ -0,0 +1,113 @@
1import unittest
2import os
3import logging
4import re
5import shutil
6import datetime
7
8import oeqa.utils.ftools as ftools
9from oeqa.selftest.base import oeSelfTest
10from oeqa.utils.commands import runCmd, bitbake, get_bb_var
11
12class BitbakePrTests(oeSelfTest):
13
14 def get_pr_version(self, package_name):
15 pkgdata_dir = get_bb_var('PKGDATA_DIR')
16 package_data_file = os.path.join(pkgdata_dir, 'runtime', package_name)
17 package_data = ftools.read_file(package_data_file)
18 find_pr = re.search("PKGR: r[0-9]+\.([0-9]+)", package_data)
19 self.assertTrue(find_pr)
20 return int(find_pr.group(1))
21
22 def get_task_stamp(self, package_name, recipe_task):
23 stampdata = get_bb_var('STAMP', target=package_name).split('/')
24 prefix = stampdata[-1]
25 package_stamps_path = "/".join(stampdata[:-1])
26 stamps = []
27 for stamp in os.listdir(package_stamps_path):
28 find_stamp = re.match("%s\.%s\.([a-z0-9]{32})" % (prefix, recipe_task), stamp)
29 if find_stamp:
30 stamps.append(find_stamp.group(1))
31 self.assertFalse(len(stamps) == 0, msg="Cound not find stamp for task %s for recipe %s" % (recipe_task, package_name))
32 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))
33 return str(stamps[0])
34
35 def increment_package_pr(self, package_name):
36 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()
37 self.write_recipeinc(package_name, inc_data)
38 bitbake("-ccleansstate %s" % package_name)
39 res = bitbake(package_name, ignore_status=True)
40 self.delete_recipeinc(package_name)
41 self.assertEqual(res.status, 0, msg=res.output)
42 self.assertTrue("NOTE: Started PRServer with DBfile" in res.output, msg=res.output)
43
44 def config_pr_tests(self, package_name, package_type='rpm', pr_socket='localhost:0'):
45 config_package_data = 'PACKAGE_CLASSES = "package_%s"' % package_type
46 self.write_config(config_package_data)
47 config_server_data = 'PRSERV_HOST = "%s"' % pr_socket
48 self.append_config(config_server_data)
49
50 def run_test_pr_service(self, package_name, package_type='rpm', track_task='do_package', pr_socket='localhost:0'):
51 self.config_pr_tests(package_name, package_type, pr_socket)
52
53 self.increment_package_pr(package_name)
54 pr_1 = self.get_pr_version(package_name)
55 stamp_1 = self.get_task_stamp(package_name, track_task)
56
57 self.increment_package_pr(package_name)
58 pr_2 = self.get_pr_version(package_name)
59 stamp_2 = self.get_task_stamp(package_name, track_task)
60
61 bitbake("-ccleansstate %s" % package_name)
62 self.assertTrue(pr_2 - pr_1 == 1)
63 self.assertTrue(stamp_1 != stamp_2)
64
65 def run_test_pr_export_import(self, package_name, replace_current_db=True):
66 self.config_pr_tests(package_name)
67
68 self.increment_package_pr(package_name)
69 pr_1 = self.get_pr_version(package_name)
70
71 exported_db_path = os.path.join(self.builddir, 'export.inc')
72 export_result = runCmd("bitbake-prserv-tool export %s" % exported_db_path, ignore_status=True)
73 self.assertEqual(export_result.status, 0, msg="PR Service database export failed: %s" % export_result.output)
74
75 if replace_current_db:
76 current_db_path = os.path.join(get_bb_var('PERSISTENT_DIR'), 'prserv.sqlite3')
77 self.assertTrue(os.path.exists(current_db_path), msg="Path to current PR Service database is invalid: %s" % current_db_path)
78 os.remove(current_db_path)
79
80 import_result = runCmd("bitbake-prserv-tool import %s" % exported_db_path, ignore_status=True)
81 os.remove(exported_db_path)
82 self.assertEqual(import_result.status, 0, msg="PR Service database import failed: %s" % import_result.output)
83
84 self.increment_package_pr(package_name)
85 pr_2 = self.get_pr_version(package_name)
86
87 bitbake("-ccleansstate %s" % package_name)
88 self.assertTrue(pr_2 - pr_1 == 1)
89
90
91 def test_import_export_replace_db(self):
92 self.run_test_pr_export_import('m4')
93
94 def test_import_export_override_db(self):
95 self.run_test_pr_export_import('m4', replace_current_db=False)
96
97 def test_pr_service_rpm_arch_dep(self):
98 self.run_test_pr_service('m4', 'rpm', 'do_package')
99
100 def test_pr_service_deb_arch_dep(self):
101 self.run_test_pr_service('m4', 'deb', 'do_package')
102
103 def test_pr_service_ipk_arch_dep(self):
104 self.run_test_pr_service('m4', 'ipk', 'do_package')
105
106 def test_pr_service_rpm_arch_indep(self):
107 self.run_test_pr_service('xcursor-transparent-theme', 'rpm', 'do_package')
108
109 def test_pr_service_deb_arch_indep(self):
110 self.run_test_pr_service('xcursor-transparent-theme', 'deb', 'do_package')
111
112 def test_pr_service_ipk_arch_indep(self):
113 self.run_test_pr_service('xcursor-transparent-theme', 'ipk', 'do_package')