Restructure tests into separate modules
Move all tests to a designated subdirectory and split the modules up
so that each test module covers one module.
Further rename some misnamed Test classes and adapt the test coverage
configuration so that the new test modules are omitted in the reports.
Issue: HIC-212
Change-Id: Idfa309aad34b8adae181cc76e95b67a86231bc69
diff --git a/crashreport_stats/tests/test_stats_management_command.py b/crashreport_stats/tests/test_stats_management_command.py
new file mode 100644
index 0000000..51860e9
--- /dev/null
+++ b/crashreport_stats/tests/test_stats_management_command.py
@@ -0,0 +1,602 @@
+"""Tests for the stats management command module."""
+
+from io import StringIO
+from datetime import datetime, timedelta
+import unittest
+
+import pytz
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from crashreport_stats.models import (
+ Version,
+ VersionDaily,
+ RadioVersion,
+ RadioVersionDaily,
+ StatsMetadata,
+)
+from crashreport_stats.tests.utils import Dummy
+
+from crashreports.models import Crashreport, HeartBeat
+
+# pylint: disable=too-many-public-methods
+
+
+class StatsCommandVersionsTestCase(TestCase):
+ """Test the generation of Version stats with the stats command."""
+
+ # 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_reports_with_same_timestamp_are_counted(
+ self, report_type, counter_attribute_name, **kwargs
+ ):
+ """Validate that reports with the same timestamp are counted.
+
+ Reports from different devices but the same timestamp should be
+ counted as independent reports.
+ """
+ # Create a report
+ device1 = Dummy.create_dummy_device(user=Dummy.create_dummy_user())
+ report1 = Dummy.create_dummy_report(
+ report_type, device=device1, **kwargs
+ )
+
+ # Create a second report with the same timestamp but from another device
+ device2 = Dummy.create_dummy_device(
+ user=Dummy.create_dummy_user(username=Dummy.USERNAMES[1])
+ )
+ Dummy.create_dummy_report(
+ report_type, device=device2, date=report1.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(report1, self.unique_entry_name)
+ }
+ version = self.version_class.objects.get(**get_params)
+
+ # Assert that both reports are counted
+ self.assertEqual(getattr(version, counter_attribute_name), 2)
+
+ @unittest.skip(
+ "Duplicates are dropped based on their timestamp at the moment. This is"
+ "to be adapted so that they are dropped taking into account the device"
+ "UUID as well."
+ )
+ def test_heartbeats_with_same_timestamp_are_counted(self):
+ """Validate that heartbeats with same timestamp are counted."""
+ counter_attribute_name = "heartbeats"
+ self._assert_reports_with_same_timestamp_are_counted(
+ HeartBeat, counter_attribute_name
+ )
+
+ @unittest.skip(
+ "Duplicates are dropped based on their timestamp at the moment. This is"
+ "to be adapted so that they are dropped taking into account the device"
+ "UUID as well."
+ )
+ def test_crash_reports_with_same_timestamp_are_counted(self):
+ """Validate that crash report duplicates are ignored."""
+ counter_attribute_name = "prob_crashes"
+ for unique_entry, boot_reason in zip(
+ self.unique_entries, Crashreport.CRASH_BOOT_REASONS
+ ):
+ params = {
+ "boot_reason": boot_reason,
+ self.unique_entry_name: unique_entry,
+ }
+ self._assert_reports_with_same_timestamp_are_counted(
+ Crashreport, counter_attribute_name, **params
+ )
+
+ @unittest.skip(
+ "Duplicates are dropped based on their timestamp at the moment. This is"
+ "to be adapted so that they are dropped taking into account the device"
+ "UUID as well."
+ )
+ def test_smpl_reports_with_same_timestamp_are_counted(self):
+ """Validate that smpl report duplicates are ignored."""
+ counter_attribute_name = "smpl"
+ for unique_entry, boot_reason in zip(
+ self.unique_entries, Crashreport.SMPL_BOOT_REASONS
+ ):
+ params = {
+ "boot_reason": boot_reason,
+ self.unique_entry_name: unique_entry,
+ }
+ self._assert_reports_with_same_timestamp_are_counted(
+ Crashreport, counter_attribute_name, **params
+ )
+
+ @unittest.skip(
+ "Duplicates are dropped based on their timestamp at the moment. This is"
+ "to be adapted so that they are dropped taking into account the device"
+ "UUID as well."
+ )
+ def test_other_reports_with_same_timestamp_are_counted(self):
+ """Validate that other report duplicates are ignored."""
+ counter_attribute_name = "other"
+ params = {"boot_reason": "random boot reason"}
+ self._assert_reports_with_same_timestamp_are_counted(
+ Crashreport, counter_attribute_name, **params
+ )
+
+ 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
+ )
+
+ def _assert_older_reports_update_released_on_date(
+ self, report_type, **kwargs
+ ):
+ """Test updating of the released_on date.
+
+ Validate that the released_on date is updated once an older report is
+ sent.
+ """
+ # Create a report
+ device = Dummy.create_dummy_device(user=Dummy.create_dummy_user())
+ report = Dummy.create_dummy_report(report_type, device=device, **kwargs)
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date matches the first report date
+ self.assertEqual(version.released_on, report.date.date())
+
+ # Create a second report with the a timestamp earlier in time
+ report_2_date = report.date - timedelta(days=1)
+ Dummy.create_dummy_report(
+ report_type, device=device, date=report_2_date, **kwargs
+ )
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date matches the older report date
+ self.assertEqual(version.released_on, report_2_date.date())
+
+ def _assert_newer_reports_do_not_update_released_on_date(
+ self, report_type, **kwargs
+ ):
+ """Test updating of the released_on date.
+
+ Validate that the released_on date is not updated once a newer report is
+ sent.
+ """
+ # Create a report
+ device = Dummy.create_dummy_device(user=Dummy.create_dummy_user())
+ report = Dummy.create_dummy_report(report_type, device=device, **kwargs)
+ report_1_date = report.date.date()
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date matches the first report date
+ self.assertEqual(version.released_on, report_1_date)
+
+ # Create a second report with the a timestamp later in time
+ report_2_date = report.date + timedelta(days=1)
+ Dummy.create_dummy_report(
+ report_type, device=device, date=report_2_date, **kwargs
+ )
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date matches the older report date
+ self.assertEqual(version.released_on, report_1_date)
+
+ def test_older_heartbeat_updates_released_on_date(self):
+ """Validate that older heartbeats update the release date."""
+ self._assert_older_reports_update_released_on_date(HeartBeat)
+
+ def test_older_crash_report_updates_released_on_date(self):
+ """Validate that older crash reports update the release date."""
+ self._assert_older_reports_update_released_on_date(Crashreport)
+
+ def test_newer_heartbeat_does_not_update_released_on_date(self):
+ """Validate that newer heartbeats don't update the release date."""
+ self._assert_newer_reports_do_not_update_released_on_date(HeartBeat)
+
+ def test_newer_crash_report_does_not_update_released_on_date(self):
+ """Validate that newer crash reports don't update the release date."""
+ self._assert_newer_reports_do_not_update_released_on_date(Crashreport)
+
+ def _assert_manually_changed_released_on_date_is_not_updated(
+ self, report_type, **kwargs
+ ):
+ """Test updating of manually changed released_on dates.
+
+ Validate that a manually changed released_on date is not updated when
+ new reports are sent.
+ """
+ # Create a report
+ device = Dummy.create_dummy_device(user=Dummy.create_dummy_user())
+ report = Dummy.create_dummy_report(report_type, device=device, **kwargs)
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date matches the first report date
+ self.assertEqual(version.released_on, report.date.date())
+
+ # Create a second report with a timestamp earlier in time
+ report_2_date = report.date - timedelta(days=1)
+ Dummy.create_dummy_report(
+ report_type, device=device, date=report_2_date, **kwargs
+ )
+
+ # Manually change the released_on date
+ version_release_date = report.date + timedelta(days=1)
+ version.released_on = version_release_date
+ version.save()
+
+ # Run the command to update the database
+ call_command("stats", "update")
+
+ # Get the corresponding version instance from the database
+ version = self.version_class.objects.get(
+ **{self.unique_entry_name: getattr(report, self.unique_entry_name)}
+ )
+
+ # Assert that the released_on date still matches the date is was
+ # manually changed to
+ self.assertEqual(version.released_on, version_release_date.date())
+
+ def test_manually_changed_released_on_date_is_not_updated_by_heartbeat(
+ self
+ ):
+ """Test update of manually changed released_on date with heartbeat."""
+ self._assert_manually_changed_released_on_date_is_not_updated(HeartBeat)
+
+ def test_manually_changed_released_on_date_is_not_updated_by_crash_report(
+ self
+ ):
+ """Test update of manually changed released_on date with crashreport."""
+ self._assert_manually_changed_released_on_date_is_not_updated(
+ Crashreport
+ )
+
+
+# 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
+
+
+class CommandDebugOutputTestCase(TestCase):
+ """Test the reset and update commands debug output."""
+
+ # Additional positional arguments to pass to the commands
+ _CMD_ARGS = ["--no-color", "-v 2"]
+
+ # The stats models
+ _STATS_MODELS = [Version, VersionDaily, RadioVersion, RadioVersionDaily]
+ # The models that will generate an output
+ _ALL_MODELS = _STATS_MODELS + [StatsMetadata]
+ _COUNTER_NAMES = ["heartbeats", "crashes", "smpl", "other"]
+ _COUNTER_ACTIONS = ["created", "updated"]
+
+ def _assert_command_output_matches(self, command, number, facts, models):
+ """Validate the debug output of a command.
+
+ The debug output is matched against the facts and models given in
+ the parameters.
+ """
+ buffer = StringIO()
+ call_command("stats", command, *self._CMD_ARGS, stdout=buffer)
+ output = buffer.getvalue().splitlines()
+
+ expected_output = "{number} {model} {fact}"
+ for model in models:
+ for fact in facts:
+ self.assertIn(
+ expected_output.format(
+ number=number, model=model.__name__, fact=fact
+ ),
+ output,
+ )
+
+ def test_reset_command_on_empty_db(self):
+ """Test the reset command on an empty database.
+
+ The reset command should yield nothing on an empty database.
+ """
+ self._assert_command_output_matches(
+ "reset", 0, ["deleted"], self._ALL_MODELS
+ )
+
+ def test_update_command_on_empty_db(self):
+ """Test the update command on an empty database.
+
+ The update command should yield nothing on an empty database.
+ """
+ pattern = "{action} for counter {counter}"
+ facts = [
+ pattern.format(action=counter_action, counter=counter_name)
+ for counter_action in self._COUNTER_ACTIONS
+ for counter_name in self._COUNTER_NAMES
+ ]
+ self._assert_command_output_matches(
+ "update", 0, facts, self._STATS_MODELS
+ )
+
+ def test_reset_command_deletion_of_instances(self):
+ """Test the deletion of stats model instances with the reset command.
+
+ This test validates that model instances get deleted when the
+ reset command is called on a database that only contains a single
+ model instance for each class.
+ """
+ # Create dummy version instances
+ version = Dummy.create_dummy_version()
+ radio_version = Dummy.create_dummy_radio_version()
+ Dummy.create_dummy_daily_version(version)
+ Dummy.create_dummy_daily_radio_version(radio_version)
+ Dummy.create_dummy_stats_metadata()
+
+ # We expect that the model instances get deleted
+ self._assert_command_output_matches(
+ "reset", 1, ["deleted"], self._ALL_MODELS
+ )