[bug] Flaky test: test_topology_graph #347
Labels
bug
Something isn't working
Comments
4 tasks
4 tasks
|
Here's a quick and dirty way I used to make the flaky test fail locally. Open a terminal tab and run Apply this patch to only keep the flaky test. diff --git a/tests/runtests.py b/tests/runtests.py
index b3129f5..a522661 100644
--- a/tests/runtests.py
+++ b/tests/runtests.py
@@ -16,49 +16,49 @@ from selenium.webdriver.support.ui import WebDriverWait
from utils import TestUtilities
-class Pretest(TestUtilities, unittest.TestCase):
- """Checks to perform before tests"""
-
- def test_wait_for_services(self):
- """This test wait for services to be started up.
-
- Then checks if the openwisp-dashboard login page is reachable.
- Should be called first before calling another test.
- """
-
- isServiceReachable = False
- max_retries = self.config['services_max_retries']
- delay_retries = self.config['services_delay_retries']
- admin_login_page = f"{self.config['app_url']}/admin/login/"
- for _ in range(1, max_retries):
- try:
- # check if we can reach to admin login page
- # and the page return 200 OK status code
- if request.urlopen(admin_login_page, context=self.ctx).getcode() == 200:
- isServiceReachable = True
- break
- except (urlerror.HTTPError, OSError, ConnectionResetError):
- # if error occurred, retry to reach the admin
- # login page after delay_retries second(s)
- time.sleep(delay_retries)
- if not isServiceReachable:
- self.fail('ERROR: openwisp-dashboard login page not reachable!')
-
- # Ensure all celery workers are online
- container_id = self.docker_compose_get_container_id('celery')
- celery_container = self.docker_client.containers.get(container_id)
- for _ in range(0, max_retries):
- result = celery_container.exec_run('celery -A openwisp status')
- online_workers = result.output.decode('utf-8').split('\n')[-2]
- try:
- assert online_workers == '5 nodes online.'
- break
- except AssertionError:
- # if error occurred, retry to reach the celery workers
- # after delay_retries second(s)
- time.sleep(delay_retries)
- else:
- self.fail(f'All celery workers are not online: {online_workers}')
+# class Pretest(TestUtilities, unittest.TestCase):
+# """Checks to perform before tests"""
+#
+# def test_wait_for_services(self):
+# """This test wait for services to be started up.
+#
+# Then checks if the openwisp-dashboard login page is reachable.
+# Should be called first before calling another test.
+# """
+#
+# isServiceReachable = False
+# max_retries = self.config['services_max_retries']
+# delay_retries = self.config['services_delay_retries']
+# admin_login_page = f"{self.config['app_url']}/admin/login/"
+# for _ in range(1, max_retries):
+# try:
+# # check if we can reach to admin login page
+# # and the page return 200 OK status code
+# if request.urlopen(admin_login_page, context=self.ctx).getcode() == 200:
+# isServiceReachable = True
+# break
+# except (urlerror.HTTPError, OSError, ConnectionResetError):
+# # if error occurred, retry to reach the admin
+# # login page after delay_retries second(s)
+# time.sleep(delay_retries)
+# if not isServiceReachable:
+# self.fail('ERROR: openwisp-dashboard login page not reachable!')
+#
+# # Ensure all celery workers are online
+# container_id = self.docker_compose_get_container_id('celery')
+# celery_container = self.docker_client.containers.get(container_id)
+# for _ in range(0, max_retries):
+# result = celery_container.exec_run('celery -A openwisp status')
+# online_workers = result.output.decode('utf-8').split('\n')[-2]
+# try:
+# assert online_workers == '5 nodes online.'
+# break
+# except AssertionError:
+# # if error occurred, retry to reach the celery workers
+# # after delay_retries second(s)
+# time.sleep(delay_retries)
+# else:
+# self.fail(f'All celery workers are not online: {online_workers}')
class TestServices(TestUtilities, unittest.TestCase):
@@ -221,317 +221,317 @@ class TestServices(TestUtilities, unittest.TestCase):
self._wait_for_element()
self.assertIn('<li>Nodes: ', self.base_driver.page_source)
- def test_admin_login(self):
- self.login()
- self.login(driver=self.second_driver)
- try:
- self.base_driver.find_element(By.CLASS_NAME, 'logout')
- self.second_driver.find_element(By.CLASS_NAME, 'logout')
- except NoSuchElementException:
- message = (
- 'Login failed. Credentials used were username: '
- f"{self.config['username']} & Password: {self.config['password']}"
- )
- self.fail(message)
-
- def test_device_monitoring_charts(self):
- self.login()
- self._wait_for_element()
- self.get_resource('test-device', '/admin/config/device/')
- self._wait_for_element()
- self.base_driver.find_element(By.CSS_SELECTOR, 'ul.tabs li.charts').click()
- try:
- WebDriverWait(self.base_driver, 3).until(EC.alert_is_present())
- except TimeoutException:
- # No alert means that the request to fetch
- # monitoring charts was successful.
- pass
- else:
- # When the request to fetch monitoring charts fails,
- # an error is shown.
- self.fail('An alert was found on the device chart page.')
-
- def test_default_topology(self):
- self.login()
- self._wait_for_element()
- self.get_resource('test-device', '/admin/topology/topology/')
-
- def test_create_prefix_users(self):
- self.login()
- prefix_objname = 'automated-prefix-test-01'
- # Create prefix users
- self.base_driver.get(
- f"{self.config['app_url']}/admin/openwisp_radius/radiusbatch/add/"
- )
- self._wait_for_element()
- self.base_driver.find_element(By.NAME, 'strategy').find_element(
- By.XPATH, '//option[@value="prefix"]'
- ).click()
- self.base_driver.find_element(By.NAME, 'organization').find_element(
- By.XPATH, '//option[text()="default"]'
- ).click()
- self.base_driver.find_element(By.NAME, 'name').send_keys(prefix_objname)
- self.base_driver.find_element(By.NAME, 'prefix').send_keys('automated-prefix')
- self.base_driver.find_element(By.NAME, 'number_of_users').send_keys('1')
- self.base_driver.find_element(By.NAME, '_save').click()
- # Check PDF available
- self.get_resource(prefix_objname, '/admin/openwisp_radius/radiusbatch/')
- self._wait_for_element()
- self.objects_to_delete.append(self.base_driver.current_url)
- prefix_pdf_file_path = self.base_driver.find_element(
- By.XPATH, '//a[text()="Download User Credentials"]'
- ).get_property('href')
- reqHeader = {
- 'Cookie': f"sessionid={self.base_driver.get_cookies()[0]['value']}"
- }
- curlRequest = request.Request(prefix_pdf_file_path, headers=reqHeader)
- try:
- if request.urlopen(curlRequest, context=self.ctx).getcode() != 200:
- raise ValueError
- except (urlerror.HTTPError, OSError, ConnectionResetError, ValueError) as error:
- self.fail(f'Cannot download PDF file: {error}')
-
- def test_console_errors(self):
- url_list = [
- '/admin/',
- '/admin/geo/location/add/',
- '/accounts/password/reset/',
- '/admin/config/device/add/',
- '/admin/config/template/add/',
- '/admin/openwisp_radius/radiuscheck/add/',
- '/admin/openwisp_radius/radiusgroup/add/',
- '/admin/openwisp_radius/radiusbatch/add/',
- '/admin/openwisp_radius/nas/add/',
- '/admin/openwisp_radius/radiusreply/',
- '/admin/geo/floorplan/add/',
- '/admin/topology/link/add/',
- '/admin/topology/node/add/',
- '/admin/topology/topology/add/',
- '/admin/pki/ca/add/',
- '/admin/pki/cert/add/',
- '/admin/openwisp_users/user/add/',
- '/admin/firmware_upgrader/build/',
- '/admin/firmware_upgrader/build/add/',
- '/admin/firmware_upgrader/category/',
- '/admin/firmware_upgrader/category/add/',
- ]
- change_form_list = [
- ['automated-selenium-location01', '/admin/geo/location/'],
- ['users', '/admin/openwisp_radius/radiusgroup/'],
- ['default-management-vpn', '/admin/config/template/'],
- ['default', '/admin/config/vpn/'],
- ['default', '/admin/pki/ca/'],
- ['default', '/admin/pki/cert/'],
- ['default', '/admin/openwisp_users/organization/'],
- ['test_superuser2', '/admin/openwisp_users/user/'],
- ]
- self.login()
- self.create_mobile_location('automated-selenium-location01')
- self.create_superuser('[email protected]', 'test_superuser2')
- # url_list tests
- for url in url_list:
- self.base_driver.get(f"{self.config['app_url']}{url}")
- self.assertEqual([], self.console_error_check())
- self.assertIn('OpenWISP', self.base_driver.title)
- # change_form_list tests
- for change_form in change_form_list:
- self.get_resource(change_form[0], change_form[1])
- self.assertEqual([], self.console_error_check())
- self.assertIn('OpenWISP', self.base_driver.title)
-
- def test_websocket_marker(self):
- """Ensures that the websocket service is running correctly.
-
- This test uses selenium, it creates a new location, sets a map
- marker and checks if the location changed int a second window.
- """
- location_name = 'automated-websocket-selenium-loc01'
- self.login()
- self.login(driver=self.second_driver)
- self.create_mobile_location(location_name)
- self.get_resource(location_name, '/admin/geo/location/')
- self.get_resource(
- location_name, '/admin/geo/location/', driver=self.second_driver
- )
- self.base_driver.find_element(By.NAME, 'is_mobile').click()
- mark = len(self.base_driver.find_elements(By.CLASS_NAME, 'leaflet-marker-icon'))
- self.assertEqual(mark, 0)
- self.add_mobile_location_point(location_name, driver=self.second_driver)
- mark = len(self.base_driver.find_elements(By.CLASS_NAME, 'leaflet-marker-icon'))
- self.assertEqual(mark, 1)
-
- def test_add_superuser(self):
- """Create new user to ensure a new user can be added."""
- self.login()
- self.create_superuser()
- self.assertEqual(
- 'The user “test_superuser” was changed successfully.',
- self.base_driver.find_element(By.CLASS_NAME, 'success').text,
- )
-
- def test_forgot_password(self):
- """Test forgot password to ensure that postfix is working properly."""
- self.base_driver.get(f"{self.config['app_url']}/accounts/password/reset/")
- self.base_driver.find_element(By.NAME, 'email').send_keys('[email protected]')
- self.base_driver.find_element(By.XPATH, '//button[@type="submit"]').click()
- self._wait_for_element()
- self.assertIn(
- 'We have sent you an email. If you have not received '
- 'it please check your spam folder. Otherwise contact us '
- 'if you do not receive it in a few minutes.',
- self.base_driver.page_source,
- )
-
- def test_celery(self):
- """Ensure celery and celery-beat tasks are registered."""
- expected_output_list = [
- "djcelery_email_send_multiple",
- "openwisp.tasks.radius_tasks",
- "openwisp.tasks.save_snapshot",
- "openwisp.tasks.update_topology",
- "openwisp_controller.config.tasks.change_devices_templates",
- "openwisp_controller.config.tasks.create_vpn_dh",
- "openwisp_controller.config.tasks.invalidate_devicegroup_cache_change",
- "openwisp_controller.config.tasks.invalidate_devicegroup_cache_delete",
- "openwisp_controller.config.tasks.invalidate_vpn_server_devices_cache_change", # noqa: E501
- "openwisp_controller.config.tasks.trigger_vpn_server_endpoint",
- "openwisp_controller.config.tasks.update_template_related_config_status",
- "openwisp_controller.connection.tasks.auto_add_credentials_to_devices",
- "openwisp_controller.connection.tasks.launch_command",
- "openwisp_controller.connection.tasks.update_config",
- "openwisp_controller.subnet_division.tasks.provision_extra_ips",
- "openwisp_controller.subnet_division.tasks.provision_subnet_ip_for_existing_devices", # noqa: E501
- "openwisp_controller.subnet_division.tasks.update_subnet_division_index",
- "openwisp_controller.subnet_division.tasks.update_subnet_name_description",
- "openwisp_firmware_upgrader.tasks.batch_upgrade_operation",
- "openwisp_firmware_upgrader.tasks.create_all_device_firmwares",
- "openwisp_firmware_upgrader.tasks.create_device_firmware",
- "openwisp_firmware_upgrader.tasks.upgrade_firmware",
- "openwisp_monitoring.check.tasks.auto_create_config_check",
- "openwisp_monitoring.check.tasks.auto_create_iperf3_check",
- "openwisp_monitoring.check.tasks.auto_create_ping",
- "openwisp_monitoring.check.tasks.perform_check",
- "openwisp_monitoring.check.tasks.run_checks",
- "openwisp_monitoring.device.tasks.delete_wifi_clients_and_sessions",
- "openwisp_monitoring.device.tasks.offline_device_close_session",
- "openwisp_monitoring.device.tasks.trigger_device_checks",
- "openwisp_monitoring.device.tasks.write_device_metrics",
- "openwisp_monitoring.device.tasks.handle_disabled_organization",
- "openwisp_monitoring.monitoring.tasks.delete_timeseries",
- "openwisp_monitoring.monitoring.tasks.migrate_timeseries_database",
- "openwisp_monitoring.monitoring.tasks.timeseries_batch_write",
- "openwisp_monitoring.monitoring.tasks.timeseries_write",
- "openwisp_monitoring.monitoring.tasks.delete_timeseries",
- "openwisp_notifications.tasks.delete_ignore_object_notification",
- "openwisp_notifications.tasks.delete_notification",
- "openwisp_notifications.tasks.delete_obsolete_objects",
- "openwisp_notifications.tasks.delete_old_notifications",
- "openwisp_notifications.tasks.ns_organization_created",
- "openwisp_notifications.tasks.ns_organization_user_deleted",
- "openwisp_notifications.tasks.ns_register_unregister_notification_type",
- "openwisp_notifications.tasks.update_org_user_notificationsetting",
- "openwisp_notifications.tasks.update_superuser_notification_settings",
- "openwisp_radius.tasks.cleanup_stale_radacct",
- "openwisp_radius.tasks.convert_called_station_id",
- "openwisp_radius.tasks.deactivate_expired_users",
- "openwisp_radius.tasks.delete_old_postauth",
- "openwisp_radius.tasks.delete_old_radacct",
- "openwisp_radius.tasks.delete_old_radiusbatch_users",
- "openwisp_radius.tasks.delete_unverified_users",
- "openwisp_radius.tasks.perform_change_of_authorization",
- "openwisp_radius.tasks.send_login_email",
- ]
-
- def _test_celery_task_registered(container_name):
- container_id = self.docker_compose_get_container_id(container_name)
- celery_container = self.docker_client.containers.get(container_id)
- result = celery_container.exec_run('celery -A openwisp inspect registered')
- self.assertEqual(result.exit_code, 0)
-
- output = result.output.decode('utf-8')
- for expected_output in expected_output_list:
- if expected_output not in output:
- self.fail(
- 'Not all celery / celery-beat tasks are registered.\n'
- f'Expected celery task not found:\n{expected_output}'
- )
-
- with self.subTest('Test celery container'):
- _test_celery_task_registered('celery')
-
- with self.subTest('Test celery_monitoring container'):
- _test_celery_task_registered('celery_monitoring')
-
- def test_radius_user_registration(self):
- """Ensure users can register using the RADIUS API."""
- url = f'{self.config["api_url"]}/api/v1/radius/organization/default/account/'
- response = requests.post(
- url,
- json={
- 'username': 'signup-user',
- 'email': '[email protected]',
- 'password1': 'rLx6OH%[',
- 'password2': 'rLx6OH%[',
- },
- verify=False,
- )
- self.assertEqual(response.status_code, 201)
- # Delete the created user
- self.login()
- self.get_resource(
- 'signup-user', '/admin/openwisp_users/user/', 'field-username'
- )
- self.objects_to_delete.append(self.base_driver.current_url)
-
- def test_freeradius(self):
- """Ensure freeradius service is working correctly."""
- token_page = (
- f"{self.config['api_url']}/api/v1/radius/"
- "organization/default/account/token/"
- )
- request_body = "username=admin&password=admin".encode('utf-8')
- request_info = request.Request(token_page, data=request_body)
- try:
- response = request.urlopen(request_info, context=self.ctx)
- except (urlerror.HTTPError, OSError, ConnectionResetError):
- self.fail(f"Couldn't get radius-token, check {self.config['api_url']}")
- self.assertIn('"is_active":true', response.read().decode())
-
- container_id = self.docker_compose_get_container_id('freeradius')
- freeradius_container = self.docker_client.containers.get(container_id)
- freeradius_container.exec_run('apk add freeradius freeradius-radclient')
- result = freeradius_container.exec_run(
- 'radtest admin admin localhost 0 testing123'
- )
- self.assertEqual(result.exit_code, 0)
- self.assertIn('Received Access-Accept', result.output.decode('utf-8'))
-
- remove_tainted_container = [
- 'docker compose rm -sf freeradius',
- 'docker compose up -d freeradius',
- ]
- for command in remove_tainted_container:
- subprocess.Popen(
- command.split(),
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
- cwd=self.root_location,
- ).communicate()
-
- def test_containers_down(self):
- """Ensure freeradius service is working correctly."""
- cmd = subprocess.Popen(
- ['docker', 'compose', 'ps'],
- universal_newlines=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- cwd=self.root_location,
- )
- output, error = map(str, cmd.communicate())
- if 'Exit' in output:
- self.fail(
- f'One of the containers are down!\nOutput:\n{output}\nError:\n{error}'
- )
+ # def test_admin_login(self):
+ # self.login()
+ # self.login(driver=self.second_driver)
+ # try:
+ # self.base_driver.find_element(By.CLASS_NAME, 'logout')
+ # self.second_driver.find_element(By.CLASS_NAME, 'logout')
+ # except NoSuchElementException:
+ # message = (
+ # 'Login failed. Credentials used were username: '
+ # f"{self.config['username']} & Password: {self.config['password']}"
+ # )
+ # self.fail(message)
+ #
+ # def test_device_monitoring_charts(self):
+ # self.login()
+ # self._wait_for_element()
+ # self.get_resource('test-device', '/admin/config/device/')
+ # self._wait_for_element()
+ # self.base_driver.find_element(By.CSS_SELECTOR, 'ul.tabs li.charts').click()
+ # try:
+ # WebDriverWait(self.base_driver, 3).until(EC.alert_is_present())
+ # except TimeoutException:
+ # # No alert means that the request to fetch
+ # # monitoring charts was successful.
+ # pass
+ # else:
+ # # When the request to fetch monitoring charts fails,
+ # # an error is shown.
+ # self.fail('An alert was found on the device chart page.')
+ #
+ # def test_default_topology(self):
+ # self.login()
+ # self._wait_for_element()
+ # self.get_resource('test-device', '/admin/topology/topology/')
+ #
+ # def test_create_prefix_users(self):
+ # self.login()
+ # prefix_objname = 'automated-prefix-test-01'
+ # # Create prefix users
+ # self.base_driver.get(
+ # f"{self.config['app_url']}/admin/openwisp_radius/radiusbatch/add/"
+ # )
+ # self._wait_for_element()
+ # self.base_driver.find_element(By.NAME, 'strategy').find_element(
+ # By.XPATH, '//option[@value="prefix"]'
+ # ).click()
+ # self.base_driver.find_element(By.NAME, 'organization').find_element(
+ # By.XPATH, '//option[text()="default"]'
+ # ).click()
+ # self.base_driver.find_element(By.NAME, 'name').send_keys(prefix_objname)
+ # self.base_driver.find_element(By.NAME, 'prefix').send_keys('automated-prefix')
+ # self.base_driver.find_element(By.NAME, 'number_of_users').send_keys('1')
+ # self.base_driver.find_element(By.NAME, '_save').click()
+ # # Check PDF available
+ # self.get_resource(prefix_objname, '/admin/openwisp_radius/radiusbatch/')
+ # self._wait_for_element()
+ # self.objects_to_delete.append(self.base_driver.current_url)
+ # prefix_pdf_file_path = self.base_driver.find_element(
+ # By.XPATH, '//a[text()="Download User Credentials"]'
+ # ).get_property('href')
+ # reqHeader = {
+ # 'Cookie': f"sessionid={self.base_driver.get_cookies()[0]['value']}"
+ # }
+ # curlRequest = request.Request(prefix_pdf_file_path, headers=reqHeader)
+ # try:
+ # if request.urlopen(curlRequest, context=self.ctx).getcode() != 200:
+ # raise ValueError
+ # except (urlerror.HTTPError, OSError, ConnectionResetError, ValueError) as error:
+ # self.fail(f'Cannot download PDF file: {error}')
+ #
+ # def test_console_errors(self):
+ # url_list = [
+ # '/admin/',
+ # '/admin/geo/location/add/',
+ # '/accounts/password/reset/',
+ # '/admin/config/device/add/',
+ # '/admin/config/template/add/',
+ # '/admin/openwisp_radius/radiuscheck/add/',
+ # '/admin/openwisp_radius/radiusgroup/add/',
+ # '/admin/openwisp_radius/radiusbatch/add/',
+ # '/admin/openwisp_radius/nas/add/',
+ # '/admin/openwisp_radius/radiusreply/',
+ # '/admin/geo/floorplan/add/',
+ # '/admin/topology/link/add/',
+ # '/admin/topology/node/add/',
+ # '/admin/topology/topology/add/',
+ # '/admin/pki/ca/add/',
+ # '/admin/pki/cert/add/',
+ # '/admin/openwisp_users/user/add/',
+ # '/admin/firmware_upgrader/build/',
+ # '/admin/firmware_upgrader/build/add/',
+ # '/admin/firmware_upgrader/category/',
+ # '/admin/firmware_upgrader/category/add/',
+ # ]
+ # change_form_list = [
+ # ['automated-selenium-location01', '/admin/geo/location/'],
+ # ['users', '/admin/openwisp_radius/radiusgroup/'],
+ # ['default-management-vpn', '/admin/config/template/'],
+ # ['default', '/admin/config/vpn/'],
+ # ['default', '/admin/pki/ca/'],
+ # ['default', '/admin/pki/cert/'],
+ # ['default', '/admin/openwisp_users/organization/'],
+ # ['test_superuser2', '/admin/openwisp_users/user/'],
+ # ]
+ # self.login()
+ # self.create_mobile_location('automated-selenium-location01')
+ # self.create_superuser('[email protected]', 'test_superuser2')
+ # # url_list tests
+ # for url in url_list:
+ # self.base_driver.get(f"{self.config['app_url']}{url}")
+ # self.assertEqual([], self.console_error_check())
+ # self.assertIn('OpenWISP', self.base_driver.title)
+ # # change_form_list tests
+ # for change_form in change_form_list:
+ # self.get_resource(change_form[0], change_form[1])
+ # self.assertEqual([], self.console_error_check())
+ # self.assertIn('OpenWISP', self.base_driver.title)
+ #
+ # def test_websocket_marker(self):
+ # """Ensures that the websocket service is running correctly.
+ #
+ # This test uses selenium, it creates a new location, sets a map
+ # marker and checks if the location changed int a second window.
+ # """
+ # location_name = 'automated-websocket-selenium-loc01'
+ # self.login()
+ # self.login(driver=self.second_driver)
+ # self.create_mobile_location(location_name)
+ # self.get_resource(location_name, '/admin/geo/location/')
+ # self.get_resource(
+ # location_name, '/admin/geo/location/', driver=self.second_driver
+ # )
+ # self.base_driver.find_element(By.NAME, 'is_mobile').click()
+ # mark = len(self.base_driver.find_elements(By.CLASS_NAME, 'leaflet-marker-icon'))
+ # self.assertEqual(mark, 0)
+ # self.add_mobile_location_point(location_name, driver=self.second_driver)
+ # mark = len(self.base_driver.find_elements(By.CLASS_NAME, 'leaflet-marker-icon'))
+ # self.assertEqual(mark, 1)
+ #
+ # def test_add_superuser(self):
+ # """Create new user to ensure a new user can be added."""
+ # self.login()
+ # self.create_superuser()
+ # self.assertEqual(
+ # 'The user “test_superuser” was changed successfully.',
+ # self.base_driver.find_element(By.CLASS_NAME, 'success').text,
+ # )
+ #
+ # def test_forgot_password(self):
+ # """Test forgot password to ensure that postfix is working properly."""
+ # self.base_driver.get(f"{self.config['app_url']}/accounts/password/reset/")
+ # self.base_driver.find_element(By.NAME, 'email').send_keys('[email protected]')
+ # self.base_driver.find_element(By.XPATH, '//button[@type="submit"]').click()
+ # self._wait_for_element()
+ # self.assertIn(
+ # 'We have sent you an email. If you have not received '
+ # 'it please check your spam folder. Otherwise contact us '
+ # 'if you do not receive it in a few minutes.',
+ # self.base_driver.page_source,
+ # )
+ #
+ # def test_celery(self):
+ # """Ensure celery and celery-beat tasks are registered."""
+ # expected_output_list = [
+ # "djcelery_email_send_multiple",
+ # "openwisp.tasks.radius_tasks",
+ # "openwisp.tasks.save_snapshot",
+ # "openwisp.tasks.update_topology",
+ # "openwisp_controller.config.tasks.change_devices_templates",
+ # "openwisp_controller.config.tasks.create_vpn_dh",
+ # "openwisp_controller.config.tasks.invalidate_devicegroup_cache_change",
+ # "openwisp_controller.config.tasks.invalidate_devicegroup_cache_delete",
+ # "openwisp_controller.config.tasks.invalidate_vpn_server_devices_cache_change", # noqa: E501
+ # "openwisp_controller.config.tasks.trigger_vpn_server_endpoint",
+ # "openwisp_controller.config.tasks.update_template_related_config_status",
+ # "openwisp_controller.connection.tasks.auto_add_credentials_to_devices",
+ # "openwisp_controller.connection.tasks.launch_command",
+ # "openwisp_controller.connection.tasks.update_config",
+ # "openwisp_controller.subnet_division.tasks.provision_extra_ips",
+ # "openwisp_controller.subnet_division.tasks.provision_subnet_ip_for_existing_devices", # noqa: E501
+ # "openwisp_controller.subnet_division.tasks.update_subnet_division_index",
+ # "openwisp_controller.subnet_division.tasks.update_subnet_name_description",
+ # "openwisp_firmware_upgrader.tasks.batch_upgrade_operation",
+ # "openwisp_firmware_upgrader.tasks.create_all_device_firmwares",
+ # "openwisp_firmware_upgrader.tasks.create_device_firmware",
+ # "openwisp_firmware_upgrader.tasks.upgrade_firmware",
+ # "openwisp_monitoring.check.tasks.auto_create_config_check",
+ # "openwisp_monitoring.check.tasks.auto_create_iperf3_check",
+ # "openwisp_monitoring.check.tasks.auto_create_ping",
+ # "openwisp_monitoring.check.tasks.perform_check",
+ # "openwisp_monitoring.check.tasks.run_checks",
+ # "openwisp_monitoring.device.tasks.delete_wifi_clients_and_sessions",
+ # "openwisp_monitoring.device.tasks.offline_device_close_session",
+ # "openwisp_monitoring.device.tasks.trigger_device_checks",
+ # "openwisp_monitoring.device.tasks.write_device_metrics",
+ # "openwisp_monitoring.device.tasks.handle_disabled_organization",
+ # "openwisp_monitoring.monitoring.tasks.delete_timeseries",
+ # "openwisp_monitoring.monitoring.tasks.migrate_timeseries_database",
+ # "openwisp_monitoring.monitoring.tasks.timeseries_batch_write",
+ # "openwisp_monitoring.monitoring.tasks.timeseries_write",
+ # "openwisp_monitoring.monitoring.tasks.delete_timeseries",
+ # "openwisp_notifications.tasks.delete_ignore_object_notification",
+ # "openwisp_notifications.tasks.delete_notification",
+ # "openwisp_notifications.tasks.delete_obsolete_objects",
+ # "openwisp_notifications.tasks.delete_old_notifications",
+ # "openwisp_notifications.tasks.ns_organization_created",
+ # "openwisp_notifications.tasks.ns_organization_user_deleted",
+ # "openwisp_notifications.tasks.ns_register_unregister_notification_type",
+ # "openwisp_notifications.tasks.update_org_user_notificationsetting",
+ # "openwisp_notifications.tasks.update_superuser_notification_settings",
+ # "openwisp_radius.tasks.cleanup_stale_radacct",
+ # "openwisp_radius.tasks.convert_called_station_id",
+ # "openwisp_radius.tasks.deactivate_expired_users",
+ # "openwisp_radius.tasks.delete_old_postauth",
+ # "openwisp_radius.tasks.delete_old_radacct",
+ # "openwisp_radius.tasks.delete_old_radiusbatch_users",
+ # "openwisp_radius.tasks.delete_unverified_users",
+ # "openwisp_radius.tasks.perform_change_of_authorization",
+ # "openwisp_radius.tasks.send_login_email",
+ # ]
+ #
+ # def _test_celery_task_registered(container_name):
+ # container_id = self.docker_compose_get_container_id(container_name)
+ # celery_container = self.docker_client.containers.get(container_id)
+ # result = celery_container.exec_run('celery -A openwisp inspect registered')
+ # self.assertEqual(result.exit_code, 0)
+ #
+ # output = result.output.decode('utf-8')
+ # for expected_output in expected_output_list:
+ # if expected_output not in output:
+ # self.fail(
+ # 'Not all celery / celery-beat tasks are registered.\n'
+ # f'Expected celery task not found:\n{expected_output}'
+ # )
+ #
+ # with self.subTest('Test celery container'):
+ # _test_celery_task_registered('celery')
+ #
+ # with self.subTest('Test celery_monitoring container'):
+ # _test_celery_task_registered('celery_monitoring')
+ #
+ # def test_radius_user_registration(self):
+ # """Ensure users can register using the RADIUS API."""
+ # url = f'{self.config["api_url"]}/api/v1/radius/organization/default/account/'
+ # response = requests.post(
+ # url,
+ # json={
+ # 'username': 'signup-user',
+ # 'email': '[email protected]',
+ # 'password1': 'rLx6OH%[',
+ # 'password2': 'rLx6OH%[',
+ # },
+ # verify=False,
+ # )
+ # self.assertEqual(response.status_code, 201)
+ # # Delete the created user
+ # self.login()
+ # self.get_resource(
+ # 'signup-user', '/admin/openwisp_users/user/', 'field-username'
+ # )
+ # self.objects_to_delete.append(self.base_driver.current_url)
+ #
+ # def test_freeradius(self):
+ # """Ensure freeradius service is working correctly."""
+ # token_page = (
+ # f"{self.config['api_url']}/api/v1/radius/"
+ # "organization/default/account/token/"
+ # )
+ # request_body = "username=admin&password=admin".encode('utf-8')
+ # request_info = request.Request(token_page, data=request_body)
+ # try:
+ # response = request.urlopen(request_info, context=self.ctx)
+ # except (urlerror.HTTPError, OSError, ConnectionResetError):
+ # self.fail(f"Couldn't get radius-token, check {self.config['api_url']}")
+ # self.assertIn('"is_active":true', response.read().decode())
+ #
+ # container_id = self.docker_compose_get_container_id('freeradius')
+ # freeradius_container = self.docker_client.containers.get(container_id)
+ # freeradius_container.exec_run('apk add freeradius freeradius-radclient')
+ # result = freeradius_container.exec_run(
+ # 'radtest admin admin localhost 0 testing123'
+ # )
+ # self.assertEqual(result.exit_code, 0)
+ # self.assertIn('Received Access-Accept', result.output.decode('utf-8'))
+ #
+ # remove_tainted_container = [
+ # 'docker compose rm -sf freeradius',
+ # 'docker compose up -d freeradius',
+ # ]
+ # for command in remove_tainted_container:
+ # subprocess.Popen(
+ # command.split(),
+ # stdout=subprocess.DEVNULL,
+ # stderr=subprocess.DEVNULL,
+ # cwd=self.root_location,
+ # ).communicate()
+ #
+ # def test_containers_down(self):
+ # """Ensure freeradius service is working correctly."""
+ # cmd = subprocess.Popen(
+ # ['docker', 'compose', 'ps'],
+ # universal_newlines=True,
+ # stdout=subprocess.PIPE,
+ # stderr=subprocess.PIPE,
+ # cwd=self.root_location,
+ # )
+ # output, error = map(str, cmd.communicate())
+ # if 'Exit' in output:
+ # self.fail(
+ # f'One of the containers are down!\nOutput:\n{output}\nError:\n{error}'
+ # )
if __name__ == '__main__':Open another terminal window and run the test in a loop: |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Describe the bug
We have another flaky test:
Steps To Reproduce
Run
make runtests. The test fails randomly.Expected behavior
Test should always pass.
System Informatioon:
Observed in CI builds.
The text was updated successfully, but these errors were encountered: