| """Test crashreport_stats models and the 'stats' command.""" |
| from datetime import datetime, date, timedelta |
| import pytz |
| |
| from django.core.management import call_command |
| from django.test import TestCase |
| from django.urls import reverse |
| from django.utils.http import urlencode |
| |
| from rest_framework import status |
| from rest_framework.test import APITestCase, APIClient |
| |
| from crashreport_stats.models import ( |
| Version, VersionDaily, RadioVersion, RadioVersionDaily |
| ) |
| |
| from crashreports.models import User, Device, Crashreport, HeartBeat |
| |
| |
| class Dummy(): |
| """Class for creating dummy instances for testing.""" |
| |
| # Valid unique entries |
| BUILD_FINGERPRINTS = [( |
| 'Fairphone/FP2/FP2:5.1/FP2/r4275.1_FP2_gms76_1.13.0:user/release-keys' |
| ), ( |
| 'Fairphone/FP2/FP2:5.1.1/FP2-gms75.1.13.0/FP2-gms75.1.13.0' |
| ':user/release-keys' |
| ), ( |
| 'Fairphone/FP2/FP2:6.0.1/FP2-gms-18.04.1/FP2-gms-18.04.1' |
| ':user/release-keys' |
| ), ( |
| 'Fairphone/FP2/FP2:7.1.2/18.07.2/gms-7480c31d' |
| ':user/release-keys' |
| )] |
| RADIO_VERSIONS = ['4437.1-FP2-0-07', '4437.1-FP2-0-08', |
| '4437.1-FP2-0-09', '4437.1-FP2-0-10'] |
| |
| DATES = [date(2018, 3, 19), date(2018, 3, 26), date(2018, 5, 1)] |
| |
| DEFAULT_DUMMY_VERSION_VALUES = { |
| 'build_fingerprint': BUILD_FINGERPRINTS[0], |
| 'first_seen_on': DATES[1], |
| 'released_on': DATES[0] |
| } |
| |
| DEFAULT_DUMMY_VERSION_DAILY_VALUES = { |
| 'date': DATES[1] |
| } |
| |
| DEFAULT_DUMMY_RADIO_VERSION_VALUES = { |
| 'radio_version': RADIO_VERSIONS[0], |
| 'first_seen_on': DATES[1], |
| 'released_on': DATES[0] |
| } |
| |
| DEFAULT_DUMMY_RADIO_VERSION_DAILY_VALUES = { |
| 'date': DATES[1] |
| } |
| |
| DEFAULT_DUMMY_DEVICE_VALUES = { |
| 'board_date': datetime(2015, 12, 15, 1, 23, 45, tzinfo=pytz.utc), |
| 'chipset': 'Qualcomm MSM8974PRO-AA', |
| 'token': '64111c62d521fb4724454ca6dea27e18f93ef56e' |
| } |
| |
| DEFAULT_DUMMY_USER_VALUES = { |
| 'username': 'testuser' |
| } |
| |
| DEFAULT_DUMMY_HEARTBEAT_VALUES = { |
| 'app_version': 10100, |
| 'uptime': ( |
| 'up time: 16 days, 21:49:56, idle time: 5 days, 20:55:04, ' |
| 'sleep time: 10 days, 20:46:27'), |
| 'build_fingerprint': BUILD_FINGERPRINTS[0], |
| 'radio_version': RADIO_VERSIONS[0], |
| 'date': datetime(2018, 3, 19, tzinfo=pytz.utc), |
| } |
| |
| DEFAULT_DUMMY_CRASHREPORT_VALUES = DEFAULT_DUMMY_HEARTBEAT_VALUES.copy() |
| DEFAULT_DUMMY_CRASHREPORT_VALUES.update({ |
| 'is_fake_report': 0, |
| 'boot_reason': Crashreport.BOOT_REASON_UNKOWN, |
| 'power_on_reason': 'it was powered on', |
| 'power_off_reason': 'something happened and it went off', |
| }) |
| |
| @staticmethod |
| def update_copy(original, update): |
| """Merge fields of update into a copy of original.""" |
| data = original.copy() |
| data.update(update) |
| return data |
| |
| @staticmethod |
| def create_dummy_user(**kwargs): |
| """Create a dummy user instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created user instance. |
| |
| """ |
| entity = User(**Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_USER_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_device(user, **kwargs): |
| """Create a dummy device instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| user: The user instance that the device should relate to |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created device instance. |
| |
| """ |
| entity = Device(user=user, **Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_DEVICE_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_report(report_type, device, **kwargs): |
| """Create a dummy report instance of the given report class type. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| report_type: The class of the report type to be created. |
| user: The device instance that the heartbeat should relate to |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created report instance. |
| |
| """ |
| if report_type == HeartBeat: |
| entity = HeartBeat(device=device, **Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_HEARTBEAT_VALUES, kwargs)) |
| elif report_type == Crashreport: |
| entity = Crashreport(device=device, **Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_CRASHREPORT_VALUES, kwargs)) |
| else: |
| raise RuntimeError( |
| 'No dummy report instance can be created for {}'.format( |
| report_type.__name__)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_version(**kwargs): |
| """Create a dummy version instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created version instance. |
| |
| """ |
| entity = Version(**Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_VERSION_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_radio_version(**kwargs): |
| """Create a dummy radio version instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created radio version instance. |
| |
| """ |
| entity = RadioVersion(**Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_RADIO_VERSION_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_daily_version(version, **kwargs): |
| """Create a dummy daily version instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created daily version instance. |
| |
| """ |
| entity = VersionDaily(version=version, **Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_VERSION_DAILY_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def create_dummy_daily_radio_version(version, **kwargs): |
| """Create a dummy daily radio version instance. |
| |
| The dummy instance is created and saved to the database. |
| Args: |
| **kwargs: |
| Optional arguments to extend/overwrite the default values. |
| |
| Returns: The created daily radio version instance. |
| |
| """ |
| entity = RadioVersionDaily(version=version, **Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_RADIO_VERSION_DAILY_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| |
| class _VersionTestCase(APITestCase): |
| """Abstract class for version-related test cases to inherit from.""" |
| |
| # The attribute name characterising the unicity of a stats entry (the |
| # named identifier) |
| unique_entry_name = 'build_fingerprint' |
| # The collection of unique entries to post |
| unique_entries = Dummy.BUILD_FINGERPRINTS |
| # The URL to retrieve the stats entries from |
| endpoint_url = reverse('hiccup_stats_api_v1_versions') |
| |
| @classmethod |
| def setUpTestData(cls): # noqa: N802 |
| """Create an admin user for accessing the API. |
| |
| The APIClient that can be used to make authenticated requests to the |
| server is stored in self.admin. |
| """ |
| admin_user = User.objects.create_superuser( |
| 'somebody', 'somebody@example.com', 'thepassword') |
| cls.admin = APIClient() |
| cls.admin.force_authenticate(admin_user) |
| |
| @staticmethod |
| def _create_dummy_version(**kwargs): |
| return Dummy.create_dummy_version(**kwargs) |
| |
| def _get_with_params(self, url, params): |
| return self.admin.get('{}?{}'.format(url, urlencode(params))) |
| |
| def _assert_result_length_is(self, response, count): |
| self.assertEqual(response.status_code, status.HTTP_200_OK) |
| self.assertIn('results', response.data) |
| self.assertIn('count', response.data) |
| self.assertEqual(response.data['count'], count) |
| self.assertEqual(len(response.data['results']), count) |
| |
| def _assert_device_owner_has_no_get_access(self, entries_url): |
| # Create a user and device |
| user = Dummy.create_dummy_user() |
| device = Dummy.create_dummy_device(user=user) |
| |
| # Create authenticated client |
| user = APIClient() |
| user.credentials(HTTP_AUTHORIZATION='Token ' + device.token) |
| |
| # Try getting entries using the client |
| response = user.get(entries_url) |
| self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
| |
| def _assert_filter_result_matches(self, filter_params, expected_result): |
| # List entities with filter |
| response = self._get_with_params(self.endpoint_url, filter_params) |
| |
| # Expect only the single matching result to be returned |
| self._assert_result_length_is(response, 1) |
| self.assertEqual(response.data['results'][0][self.unique_entry_name], |
| getattr(expected_result, self.unique_entry_name)) |
| |
| |
| class VersionTestCase(_VersionTestCase): |
| """Test the Version and REST endpoint.""" |
| |
| def _create_version_entities(self): |
| versions = [ |
| self._create_dummy_version( |
| **{self.unique_entry_name: unique_entry} |
| ) |
| for unique_entry in self.unique_entries |
| ] |
| return versions |
| |
| def test_list_versions_without_authentication(self): |
| """Test listing of versions without authentication.""" |
| response = self.client.get(self.endpoint_url) |
| self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
| |
| def test_list_versions_as_device_owner(self): |
| """Test listing of versions as device owner.""" |
| self._assert_device_owner_has_no_get_access(self.endpoint_url) |
| |
| def test_list_versions_empty_database(self): |
| """Test listing of versions on an empty database.""" |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, 0) |
| |
| def test_list_versions(self): |
| """Test listing versions.""" |
| versions = self._create_version_entities() |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions)) |
| |
| def test_filter_versions_by_unique_entry_name(self): |
| """Test filtering versions by their unique entry name.""" |
| versions = self._create_version_entities() |
| response = self.admin.get(self.endpoint_url) |
| |
| # Listing all entities should return the correct result length |
| self._assert_result_length_is(response, len(versions)) |
| |
| # List entities with filter |
| filter_params = { |
| self.unique_entry_name: getattr(versions[0], |
| self.unique_entry_name) |
| } |
| self._assert_filter_result_matches(filter_params, |
| expected_result=versions[0]) |
| |
| def test_filter_versions_by_release_type(self): |
| """Test filtering versions by release type.""" |
| # Create versions for all combinations of release types |
| versions = [] |
| i = 0 |
| for is_official_release in True, False: |
| for is_beta_release in True, False: |
| versions.append(self._create_dummy_version(**{ |
| 'is_official_release': is_official_release, |
| 'is_beta_release': is_beta_release, |
| self.unique_entry_name: self.unique_entries[i] |
| })) |
| i += 1 |
| |
| # # Listing all entities should return the correct result length |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions)) |
| |
| # List each of the entities with the matching filter params |
| for version in versions: |
| filter_params = { |
| 'is_official_release': version.is_official_release, |
| 'is_beta_release': version.is_beta_release |
| } |
| self._assert_filter_result_matches(filter_params, |
| expected_result=version) |
| |
| def test_filter_versions_by_first_seen_date(self): |
| """Test filtering versions by first seen date.""" |
| versions = self._create_version_entities() |
| |
| # Set the first seen date of an entity |
| versions[0].first_seen_on = Dummy.DATES[2] |
| versions[0].save() |
| |
| # Listing all entities should return the correct result length |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions)) |
| |
| # Expect the single matching result to be returned |
| filter_params = {'first_seen_after': Dummy.DATES[2]} |
| self._assert_filter_result_matches(filter_params, |
| expected_result=versions[0]) |
| |
| |
| # pylint: disable=too-many-ancestors |
| class RadioVersionTestCase(VersionTestCase): |
| """Test the RadioVersion REST endpoint.""" |
| |
| unique_entry_name = 'radio_version' |
| unique_entries = Dummy.RADIO_VERSIONS |
| endpoint_url = reverse('hiccup_stats_api_v1_radio_versions') |
| |
| @staticmethod |
| def _create_dummy_version(**kwargs): |
| return Dummy.create_dummy_radio_version(**kwargs) |
| |
| |
| class VersionDailyTestCase(_VersionTestCase): |
| """Test the VersionDaily REST endpoint.""" |
| |
| endpoint_url = reverse('hiccup_stats_api_v1_version_daily') |
| |
| @staticmethod |
| def _create_dummy_daily_version(version, **kwargs): |
| return Dummy.create_dummy_daily_version(version, **kwargs) |
| |
| def _create_version_entities(self): |
| versions = [ |
| self._create_dummy_version( |
| **{self.unique_entry_name: unique_entry} |
| ) |
| for unique_entry in self.unique_entries |
| ] |
| versions_daily = [ |
| self._create_dummy_daily_version(version=version) |
| for version in versions |
| ] |
| return versions_daily |
| |
| def test_list_daily_versions_without_authentication(self): |
| """Test listing of daily versions without authentication.""" |
| response = self.client.get(self.endpoint_url) |
| self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
| |
| def test_list_daily_versions_as_device_owner(self): |
| """Test listing of daily versions as device owner.""" |
| self._assert_device_owner_has_no_get_access(self.endpoint_url) |
| |
| def test_list_daily_versions_empty_database(self): |
| """Test listing of daily versions on an empty database.""" |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, 0) |
| |
| def test_list_daily_versions(self): |
| """Test listing daily versions.""" |
| versions_daily = self._create_version_entities() |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions_daily)) |
| |
| def test_filter_daily_versions_by_version(self): |
| """Test filtering versions by the version they relate to.""" |
| # Create VersionDaily entities |
| versions = self._create_version_entities() |
| |
| # Listing all entities should return the correct result length |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions)) |
| |
| # List entities with filter |
| param_name = 'version__' + self.unique_entry_name |
| filter_params = { |
| param_name: getattr(versions[0].version, self.unique_entry_name) |
| } |
| self._assert_filter_result_matches(filter_params, |
| expected_result=versions[0].version) |
| |
| def test_filter_daily_versions_by_date(self): |
| """Test filtering daily versions by date.""" |
| # Create Version and VersionDaily entities |
| versions = self._create_version_entities() |
| |
| # Update the date |
| versions[0].date = Dummy.DATES[2] |
| versions[0].save() |
| |
| # Listing all entities should return the correct result length |
| response = self.admin.get(self.endpoint_url) |
| self._assert_result_length_is(response, len(versions)) |
| |
| # Expect the single matching result to be returned |
| filter_params = {'date': versions[0].date} |
| self._assert_filter_result_matches(filter_params, |
| expected_result=versions[0].version) |
| |
| |
| class RadioVersionDailyTestCase(VersionDailyTestCase): |
| """Test the RadioVersionDaily REST endpoint.""" |
| |
| unique_entry_name = 'radio_version' |
| unique_entries = Dummy.RADIO_VERSIONS |
| endpoint_url = reverse('hiccup_stats_api_v1_radio_version_daily') |
| |
| @staticmethod |
| def _create_dummy_version(**kwargs): |
| entity = RadioVersion(**Dummy.update_copy( |
| Dummy.DEFAULT_DUMMY_RADIO_VERSION_VALUES, kwargs)) |
| entity.save() |
| return entity |
| |
| @staticmethod |
| def _create_dummy_daily_version(version, **kwargs): |
| return Dummy.create_dummy_daily_radio_version(version, **kwargs) |
| |
| |
| class StatsCommandVersionsTestCase(TestCase): |
| """Test the generation of Version stats with the stats command.""" |
| |
| # FIXME: Test for false duplicates: same timestamps but different UUIDs |
| # FIXME: Test that the 'released_on' field changes or not once an older |
| # report has been sent depending on whether the field has been manually |
| # changed |
| # FIXME: Test that tests the daily version stats |
| # FIXME: Test creating stats from reports of different devices/users. |
| |
| # The class of the version type to be tested |
| version_class = Version |
| # The attribute name characterising the unicity of a stats entry (the |
| # named identifier) |
| unique_entry_name = 'build_fingerprint' |
| # The collection of unique entries to post |
| unique_entries = Dummy.BUILD_FINGERPRINTS |
| |
| def _create_reports(self, report_type, unique_entry_name, device, |
| number, **kwargs): |
| # Create reports with distinct timestamps |
| now = datetime.now(pytz.utc) |
| for i in range(number): |
| report_date = now - timedelta(milliseconds=i) |
| report_attributes = { |
| self.unique_entry_name: unique_entry_name, |
| 'device': device, |
| 'date': report_date |
| } |
| report_attributes.update(**kwargs) |
| Dummy.create_dummy_report(report_type, **report_attributes) |
| |
| def test_stats_calculation(self): |
| """Test generation of a Version instance.""" |
| user = Dummy.create_dummy_user() |
| device = Dummy.create_dummy_device(user=user) |
| heartbeat = Dummy.create_dummy_report(HeartBeat, device=device) |
| |
| # Expect that we do not have the Version before updating the stats |
| get_params = { |
| self.unique_entry_name: getattr(heartbeat, self.unique_entry_name) |
| } |
| self.assertRaises(self.version_class.DoesNotExist, |
| self.version_class.objects.get, **get_params) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| # Assume that a corresponding Version instance has been created |
| version = self.version_class.objects.get(**get_params) |
| self.assertIsNotNone(version) |
| |
| def _assert_older_report_updates_version_date(self, report_type): |
| """Validate that older reports sent later affect the version date.""" |
| user = Dummy.create_dummy_user() |
| device = Dummy.create_dummy_device(user=user) |
| report = Dummy.create_dummy_report(report_type, device=device) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| get_params = { |
| self.unique_entry_name: getattr(report, self.unique_entry_name) |
| } |
| version = self.version_class.objects.get(**get_params) |
| |
| self.assertEqual(report.date.date(), version.first_seen_on) |
| |
| # Create a new report from an earlier point in time |
| report_time_2 = report.date - timedelta(weeks=1) |
| Dummy.create_dummy_report(report_type, device=device, |
| date=report_time_2) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| # Get the same version object from before |
| version = self.version_class.objects.get(**get_params) |
| |
| # Validate that the date matches the report recently sent |
| self.assertEqual(report_time_2.date(), version.first_seen_on) |
| |
| def test_older_heartbeat_updates_version_date(self): |
| """Validate updating version date with older heartbeats.""" |
| self._assert_older_report_updates_version_date(HeartBeat) |
| |
| def test_older_crash_report_updates_version_date(self): |
| """Validate updating version date with older crash reports.""" |
| self._assert_older_report_updates_version_date(Crashreport) |
| |
| def test_entries_are_unique(self): |
| """Validate the entries' unicity and value.""" |
| # Create some reports |
| user = Dummy.create_dummy_user() |
| device = Dummy.create_dummy_device(user=user) |
| for unique_entry in self.unique_entries: |
| self._create_reports(HeartBeat, unique_entry, device, 10) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| # Check whether the correct amount of distinct versions have been |
| # created |
| versions = self.version_class.objects.all() |
| for version in versions: |
| self.assertIn(getattr(version, self.unique_entry_name), |
| self.unique_entries) |
| self.assertEqual(len(versions), len(self.unique_entries)) |
| |
| def _assert_counter_distribution_is_correct(self, report_type, numbers, |
| counter_attribute_name, |
| **kwargs): |
| """Validate a counter distribution in the database.""" |
| if len(numbers) != len(self.unique_entries): |
| raise ValueError('The length of the numbers list must match the ' |
| 'length of self.unique_entries in the test class' |
| '({} != {})'.format(len(numbers), |
| len(self.unique_entries))) |
| # Create some reports |
| user = Dummy.create_dummy_user() |
| device = Dummy.create_dummy_device(user=user) |
| for unique_entry, num in zip(self.unique_entries, numbers): |
| self._create_reports(report_type, unique_entry, device, num, |
| **kwargs) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| # Check whether the numbers of reports match |
| for version in self.version_class.objects.all(): |
| unique_entry_name = getattr(version, self.unique_entry_name) |
| num = numbers[self.unique_entries.index(unique_entry_name)] |
| self.assertEqual(num, getattr(version, counter_attribute_name)) |
| |
| def test_heartbeats_counter(self): |
| """Test the calculation of the heartbeats counter.""" |
| numbers = [10, 7, 8, 5] |
| counter_attribute_name = 'heartbeats' |
| self._assert_counter_distribution_is_correct(HeartBeat, numbers, |
| counter_attribute_name) |
| |
| def test_crash_reports_counter(self): |
| """Test the calculation of the crashreports counter.""" |
| numbers = [2, 5, 0, 3] |
| counter_attribute_name = 'prob_crashes' |
| boot_reason_param = {'boot_reason': Crashreport.BOOT_REASON_UNKOWN} |
| self._assert_counter_distribution_is_correct(Crashreport, numbers, |
| counter_attribute_name, |
| **boot_reason_param) |
| |
| def test_smpl_reports_counter(self): |
| """Test the calculation of the smpl reports counter.""" |
| numbers = [1, 3, 4, 0] |
| counter_attribute_name = 'smpl' |
| boot_reason_param = {'boot_reason': Crashreport.BOOT_REASON_RTC_ALARM} |
| self._assert_counter_distribution_is_correct(Crashreport, numbers, |
| counter_attribute_name, |
| **boot_reason_param) |
| |
| def test_other_reports_counter(self): |
| """Test the calculation of the other reports counter.""" |
| numbers = [0, 2, 1, 2] |
| counter_attribute_name = 'other' |
| boot_reason_param = {'boot_reason': "random boot reason"} |
| self._assert_counter_distribution_is_correct(Crashreport, numbers, |
| counter_attribute_name, |
| **boot_reason_param) |
| |
| def _assert_duplicates_are_ignored(self, report_type, device, |
| counter_attribute_name, **kwargs): |
| """Validate that reports with duplicate timestamps are ignored.""" |
| # Create a report |
| report = Dummy.create_dummy_report(report_type, device=device, |
| **kwargs) |
| |
| # Create a second report with the same timestamp |
| Dummy.create_dummy_report(report_type, device=device, |
| date=report.date, **kwargs) |
| |
| # Run the command to update the database |
| call_command('stats', 'update') |
| |
| # Get the corresponding version instance from the database |
| get_params = { |
| self.unique_entry_name: getattr(report, self.unique_entry_name) |
| } |
| version = self.version_class.objects.get(**get_params) |
| |
| # Assert that the report with the duplicate timestamp is not |
| # counted, i.e. only 1 report is counted. |
| self.assertEqual(getattr(version, counter_attribute_name), 1) |
| |
| def test_heartbeat_duplicates_are_ignored(self): |
| """Validate that heartbeat duplicates are ignored.""" |
| counter_attribute_name = 'heartbeats' |
| device = Dummy.create_dummy_device(user=Dummy.create_dummy_user()) |
| self._assert_duplicates_are_ignored(HeartBeat, device, |
| counter_attribute_name) |
| |
| def test_crash_report_duplicates_are_ignored(self): |
| """Validate that crash report duplicates are ignored.""" |
| counter_attribute_name = 'prob_crashes' |
| device = Dummy.create_dummy_device(user=Dummy.create_dummy_user()) |
| for i, boot_reason in enumerate(Crashreport.CRASH_BOOT_REASONS): |
| params = {'boot_reason': boot_reason, |
| self.unique_entry_name: self.unique_entries[i]} |
| self._assert_duplicates_are_ignored(Crashreport, device, |
| counter_attribute_name, |
| **params) |
| |
| def test_smpl_report_duplicates_are_ignored(self): |
| """Validate that smpl report duplicates are ignored.""" |
| counter_attribute_name = 'smpl' |
| device = Dummy.create_dummy_device(user=Dummy.create_dummy_user()) |
| for i, boot_reason in enumerate(Crashreport.SMPL_BOOT_REASONS): |
| params = {'boot_reason': boot_reason, |
| self.unique_entry_name: self.unique_entries[i]} |
| self._assert_duplicates_are_ignored(Crashreport, device, |
| counter_attribute_name, |
| **params) |
| |
| def test_other_report_duplicates_are_ignored(self): |
| """Validate that other report duplicates are ignored.""" |
| counter_attribute_name = 'other' |
| params = {'boot_reason': 'random boot reason'} |
| device = Dummy.create_dummy_device(user=Dummy.create_dummy_user()) |
| self._assert_duplicates_are_ignored(Crashreport, device, |
| counter_attribute_name, |
| **params) |
| |
| |
| # pylint: disable=too-many-ancestors |
| class StatsCommandRadioVersionsTestCase(StatsCommandVersionsTestCase): |
| """Test the generation of RadioVersion stats with the stats command.""" |
| |
| version_class = RadioVersion |
| unique_entry_name = 'radio_version' |
| unique_entries = Dummy.RADIO_VERSIONS |