prservice.py 5.5 KB

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