1# 2# SPDX-License-Identifier: MIT 3# 4 5import os 6import re 7import shutil 8import datetime 9 10import oeqa.utils.ftools as ftools 11from oeqa.selftest.case import OESelftestTestCase 12from oeqa.utils.commands import runCmd, bitbake, get_bb_var 13from oeqa.utils.network import get_free_port 14 15class BitbakePrTests(OESelftestTestCase): 16 17 @classmethod 18 def setUpClass(cls): 19 super(BitbakePrTests, cls).setUpClass() 20 cls.pkgdata_dir = get_bb_var('PKGDATA_DIR') 21 22 def get_pr_version(self, package_name): 23 package_data_file = os.path.join(self.pkgdata_dir, 'runtime', package_name) 24 package_data = ftools.read_file(package_data_file) 25 find_pr = re.search(r"PKGR: r[0-9]+\.([0-9]+)", package_data) 26 self.assertTrue(find_pr, "No PKG revision found via regex 'PKGR: r[0-9]+\.([0-9]+)' in %s" % package_data_file) 27 return int(find_pr.group(1)) 28 29 def get_task_stamp(self, package_name, recipe_task): 30 stampdata = get_bb_var('STAMP', target=package_name).split('/') 31 prefix = stampdata[-1] 32 package_stamps_path = "/".join(stampdata[:-1]) 33 stamps = [] 34 for stamp in os.listdir(package_stamps_path): 35 find_stamp = re.match(r"%s\.%s\.([a-z0-9]{32})" % (re.escape(prefix), recipe_task), stamp) 36 if find_stamp: 37 stamps.append(find_stamp.group(1)) 38 self.assertFalse(len(stamps) == 0, msg="Cound not find stamp for task %s for recipe %s" % (recipe_task, package_name)) 39 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)) 40 return str(stamps[0]) 41 42 def increment_package_pr(self, package_name): 43 inc_data = "do_package:append() {\n bb.build.exec_func('do_test_prserv', d)\n}\ndo_test_prserv() {\necho \"The current date is: %s\" > ${PKGDESTWORK}/${PN}.datestamp\n}" % datetime.datetime.now() 44 self.write_recipeinc(package_name, inc_data) 45 res = bitbake(package_name, ignore_status=True) 46 self.delete_recipeinc(package_name) 47 self.assertEqual(res.status, 0, msg=res.output) 48 49 def config_pr_tests(self, package_name, package_type='rpm', pr_socket='localhost:0'): 50 config_package_data = 'PACKAGE_CLASSES = "package_%s"' % package_type 51 self.write_config(config_package_data) 52 config_server_data = 'PRSERV_HOST = "%s"' % pr_socket 53 self.append_config(config_server_data) 54 55 def run_test_pr_service(self, package_name, package_type='rpm', track_task='do_package', pr_socket='localhost:0'): 56 self.config_pr_tests(package_name, package_type, pr_socket) 57 58 self.increment_package_pr(package_name) 59 pr_1 = self.get_pr_version(package_name) 60 stamp_1 = self.get_task_stamp(package_name, track_task) 61 62 self.increment_package_pr(package_name) 63 pr_2 = self.get_pr_version(package_name) 64 stamp_2 = self.get_task_stamp(package_name, track_task) 65 66 self.assertTrue(pr_2 - pr_1 == 1, "New PR %s did not increment as expected (from %s), difference should be 1" % (pr_2, pr_1)) 67 self.assertTrue(stamp_1 != stamp_2, "Different pkg rev. but same stamp: %s" % stamp_1) 68 69 def run_test_pr_export_import(self, package_name, replace_current_db=True): 70 self.config_pr_tests(package_name) 71 72 self.increment_package_pr(package_name) 73 pr_1 = self.get_pr_version(package_name) 74 75 exported_db_path = os.path.join(self.builddir, 'export.inc') 76 export_result = runCmd("bitbake-prserv-tool export %s" % exported_db_path, ignore_status=True) 77 self.assertEqual(export_result.status, 0, msg="PR Service database export failed: %s" % export_result.output) 78 self.assertTrue(os.path.exists(exported_db_path), msg="%s didn't exist, tool output %s" % (exported_db_path, export_result.output)) 79 80 if replace_current_db: 81 current_db_path = os.path.join(get_bb_var('PERSISTENT_DIR'), 'prserv.sqlite3') 82 self.assertTrue(os.path.exists(current_db_path), msg="Path to current PR Service database is invalid: %s" % current_db_path) 83 os.remove(current_db_path) 84 85 import_result = runCmd("bitbake-prserv-tool import %s" % exported_db_path, ignore_status=True) 86 os.remove(exported_db_path) 87 self.assertEqual(import_result.status, 0, msg="PR Service database import failed: %s" % import_result.output) 88 89 self.increment_package_pr(package_name) 90 pr_2 = self.get_pr_version(package_name) 91 92 self.assertTrue(pr_2 - pr_1 == 1, "New PR %s did not increment as expected (from %s), difference should be 1" % (pr_2, pr_1)) 93 94 def test_import_export_replace_db(self): 95 self.run_test_pr_export_import('m4') 96 97 def test_import_export_override_db(self): 98 self.run_test_pr_export_import('m4', replace_current_db=False) 99 100 def test_pr_service_rpm_arch_dep(self): 101 self.run_test_pr_service('m4', 'rpm', 'do_package') 102 103 def test_pr_service_deb_arch_dep(self): 104 self.run_test_pr_service('m4', 'deb', 'do_package') 105 106 def test_pr_service_ipk_arch_dep(self): 107 self.run_test_pr_service('m4', 'ipk', 'do_package') 108 109 def test_pr_service_rpm_arch_indep(self): 110 self.run_test_pr_service('xcursor-transparent-theme', 'rpm', 'do_package') 111 112 def test_pr_service_deb_arch_indep(self): 113 self.run_test_pr_service('xcursor-transparent-theme', 'deb', 'do_package') 114 115 def test_pr_service_ipk_arch_indep(self): 116 self.run_test_pr_service('xcursor-transparent-theme', 'ipk', 'do_package') 117 118 def test_stopping_prservice_message(self): 119 port = get_free_port() 120 121 runCmd('bitbake-prserv --host localhost --port %s --loglevel=DEBUG --start' % port) 122 ret = runCmd('bitbake-prserv --host localhost --port %s --loglevel=DEBUG --stop' % port) 123 124 self.assertEqual(ret.status, 0) 125 126