StatisticsChannelsTest.test_list() - Code Metrics - Inspection of "Bump version: 0.3.2 → 0.3.3" - Crystalnix/omaha-server - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 67eae9...f0f1ec )

unknown

created 2016-04-07 06:56 UTC

StatisticsChannelsTest.test_list() A

↳ Parent: StatisticsChannelsTest

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	6
rs	9.4285
cc	1

# coding: utf8

"""
This software is licensed under the Apache 2 license, quoted below.

Copyright 2014 Crystalnix Limited

Licensed under the Apache License, Version 2.0 (the "License"); you may not
use this file except in compliance with the License. You may obtain a copy of
the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
License for the specific language governing permissions and limitations under
the License.
"""
from __future__ import unicode_literals
from builtins import bytes, range

import base64
from datetime import datetime, timedelta
from uuid import UUID

from django.core.urlresolvers import reverse
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.conf import settings

from rest_framework import status
from rest_framework.test import APITestCase, APIClient
from bitmapist import mark_event
from freezegun import freeze_time
import pytz

from omaha_server.utils import is_private
from omaha.statistics import userid_counting, get_users_versions, get_channel_statistics
from omaha.utils import redis
from omaha.serializers import (
    AppSerializer,
    DataSerializer,
    PlatformSerializer,
    ChannelSerializer,
    VersionSerializer,
    ActionSerializer,
    StatisticsMonthsSerializer,
    ServerVersionSerializer,
)
from omaha.factories import ApplicationFactory, DataFactory, PlatformFactory, ChannelFactory, VersionFactory, ActionFactory
from omaha.models import Application, Data, Channel, Platform, Version, Action
from omaha.tests.utils import temporary_media_root
from sparkle.models import SparkleVersion

User = get_user_model()


class BaseTest(object):
    url = None
    url_args = ()
    url_detail = None
    factory = None
    serializer = None
    maxDiff = None
    is_private = True

    def _is_private(self):
        if not self.is_private and not settings.IS_PRIVATE:
            return True
        elif self.is_private and settings.IS_PRIVATE:
            return True
        else:
            return False

    def setUp(self):
        self.objects = self.factory.create_batch(10)
        self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
        self.client.credentials(
            HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())

    def test_unauthorized(self):
        if not self._is_private():
            return
        client = APIClient()
        response = client.get(reverse(self.url, args=self.url_args), format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    def test_list(self):
        if not self._is_private():
            return
        response = self.client.get(reverse(self.url, args=self.url_args), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data), 10)
        self.assertEqual(self.serializer(self.objects, many=True).data, response.data[::-1])

    def test_detail(self):
        if not self._is_private():
            return
        obj = self.objects[0]
        url = reverse(self.url_detail, kwargs=dict(pk=obj.pk))

        response = self.client.get(url, format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertDictEqual(response.data, self.serializer(obj).data)


class AppTest(BaseTest, APITestCase):
    url = 'application-list'
    url_detail = 'application-detail'
    factory = ApplicationFactory
    serializer = AppSerializer

    @is_private()
    def test_create(self):
        data = dict(id='test_id', name='test_name', data_set=[])
        response = self.client.post(reverse(self.url
                                            ), data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj = Application.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(obj).data)


class DataTest(BaseTest, APITestCase):

    url = 'data-list'
    url_detail = 'data-detail'
    factory = DataFactory
    serializer = DataSerializer

    @is_private()
    def test_create(self):
        app = ApplicationFactory.create()
        data = dict(name=0, app=app.pk)
        response = self.client.post(reverse(self.url), data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj = Data.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(obj).data)


class PlatformTest(BaseTest, APITestCase):
    url = 'platform-list'
    url_detail = 'platform-detail'
    factory = PlatformFactory
    serializer = PlatformSerializer

    @is_private()
    def test_list(self):
        super(PlatformTest, self).test_list()

    @is_private()
    def test_create(self):
        data = dict(name='test_name')
        response = self.client.post(reverse(self.url), data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj = Platform.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(obj).data)


class ChannelTest(BaseTest, APITestCase):
    url = 'channel-list'
    url_detail = 'channel-detail'
    factory = ChannelFactory
    serializer = ChannelSerializer

    @is_private()
    def test_create(self):
        data = dict(name='test_name')
        response = self.client.post(reverse(self.url), data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj = Channel.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(obj).data)


class VersionTest(BaseTest, APITestCase):
    url = 'version-list'
    url_detail = 'version-detail'
    factory = VersionFactory
    serializer = VersionSerializer

    @is_private()
    @temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
    def test_detail(self):
        super(VersionTest, self).test_detail()

    @is_private()
    @temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
    def test_list(self):
        super(VersionTest, self).test_list()

    @is_private()
    @temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
    def test_create(self):
        data = dict(
            app=ApplicationFactory.create().id,
            platform=PlatformFactory.create().id,
            channel=ChannelFactory.create().id,
            version='1.2.3.4',
            file=SimpleUploadedFile("chrome.exe", b'content'),
        )
        response = self.client.post(reverse(self.url), data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        version = Version.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(version).data)
        self.assertEqual(version.file_size, len(b'content'))
        self.assertTrue(version.is_enabled)


class ActionTest(BaseTest, APITestCase):

    url = 'action-list'
    url_detail = 'action-detail'
    factory = ActionFactory
    serializer = ActionSerializer

    @is_private()
    def test_create(self):
        version = VersionFactory.create()
        data = dict(event=1, version=version.pk)
        response = self.client.post(reverse(self.url), data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj = Action.objects.get(id=response.data['id'])
        self.assertEqual(response.data, self.serializer(obj).data)

class LiveStatistics(APITestCase):
    maxDiff = None

    def _generate_fake_statistics(self):
        # now = datetime(2016, 2, 13)
        date = datetime(2016, 2, 13, 0)
        for i in range(self.n_hours):
            for id in range(0, i):
                mark_event('online:app:win:2.0.0.0', id, now=date, track_hourly=True)
                mark_event('online:app:mac:4.0.0.1', id, now=date, track_hourly=True)
            for id in range(i, self.n_hours):
                mark_event('online:app:win:1.0.0.0', id, now=date, track_hourly=True)
                mark_event('online:app:mac:3.0.0.0', id, now=date, track_hourly=True)
            date += timedelta(hours=1)

    def setUp(self):
        self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
        self.client.credentials(
            HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())

        redis.flushdb()
        self.app = Application.objects.create(id='app', name='app')
        self.channel = Channel.objects.create(name='stable')
        self.platform = Platform.objects.create(name='win')
        self.version1 = Version.objects.create(
            app=self.app,
            platform=self.platform,
            channel=self.channel,
            version='1.0.0.0',
            file=SimpleUploadedFile('./chrome_installer.exe', False))

        self.version2 = Version.objects.create(
            app=self.app,
            platform=self.platform,
            channel=self.channel,
            version='2.0.0.0',
            file=SimpleUploadedFile('./chrome_installer.exe', False))

        self.sparkle_version1 = SparkleVersion.objects.create(
            app=self.app,
            channel=self.channel,
            version='0.0',
            short_version='3.0.0.0',
            file=SimpleUploadedFile('./chrome_installer.dmg', False))

        self.sparkle_version2 = SparkleVersion.objects.create(
            app=self.app,
            channel=self.channel,
            version='0.1',
            short_version='4.0.0.1',
            file=SimpleUploadedFile('./chrome_installer.dmg', False))

        self.n_hours = 36
        self._generate_fake_statistics()

        hours = [datetime(2016, 2, 13, 0, tzinfo=pytz.UTC) + timedelta(hours=hour)
                 for hour in range(self.n_hours)]
        self.win_statistics = [('1.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
                                            for (i, hour)in enumerate(hours)])]
        self.win_statistics.append(('2.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
                                                for (i, hour)in enumerate(hours)]))

        self.mac_statistics = [('3.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
                                            for (i, hour)in enumerate(hours)])]
        self.mac_statistics.append(('4.0.0.1', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
                                                for (i, hour)in enumerate(hours)]))

        self.data = dict(data=dict(win=dict(self.win_statistics),
                                   mac=dict(self.mac_statistics)))

    @is_private()
    def test_unauthorized(self):
        client = APIClient()
        response = client.get(reverse('api-statistics-live', args=('app',)), format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    @is_private()
    def test_list(self):
        start = datetime(2016, 2, 13, 0)
        end = start + timedelta(hours=self.n_hours-1)
        response = self.client.get(reverse('api-statistics-live', args=('app',)),
                                   dict(start=start.isoformat(), end=end.isoformat()),
                                   format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertDictEqual(StatisticsMonthsSerializer(self.data).data, response.data)



class StatisticsMonthsMixin(object):
    url = None
    url_args = ()
    serializer = None
    url_get_params = None

    def _generate_fake_statistics(self):
        now = datetime.now()
        prev_year = now.year - 1

        for i in range(1, 13):
            date = datetime(year=prev_year, month=i, day=10)
            for id in range(1, i + 1):
                user_id = UUID(int=id)
                userid_counting(user_id, self.app_list, self.platform.name, now=date)
                user_id = UUID(int=1000 + id)
                userid_counting(user_id, [self.mac_app], 'mac', now=date)

        user_id = UUID(int=13)
        userid_counting(user_id, self.app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
        user_id = UUID(int=1013)
        userid_counting(user_id, [self.mac_app], 'mac', now=datetime(year=now.year, month=1, day=1))

    @freeze_time("2016-01-27")
    @temporary_media_root()
    def setUp(self):
        self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
        self.client.credentials(
            HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())

        redis.flushdb()
        self.app = Application.objects.create(id='app', name='app')
        self.channel = Channel.objects.create(name='stable')
        self.platform = Platform.objects.create(name='win')
        self.version1 = Version.objects.create(
            app=self.app,
            platform=self.platform,
            channel=self.channel,
            version='1.0.0.0',
            file=SimpleUploadedFile('./chrome_installer.exe', False))
        self.version2 = Version.objects.create(
            app=self.app,
            platform=self.platform,
            channel=self.channel,
            version='2.0.0.0',
            file=SimpleUploadedFile('./chrome_installer.exe', False))
        self.mac_version = SparkleVersion.objects.create(
            app=self.app,
            channel=self.channel,
            version='782.112',
            short_version='13.0.782.112',
            dsa_signature='MCwCFCdoW13VBGJWIfIklKxQVyetgxE7AhQTVuY9uQT0KOV1UEk21epBsGZMPg==',
            file=SimpleUploadedFile('./chrome.dmg', b'_' * 1024),
            file_size=1024)
        self.app_list = [dict(appid=self.app.id, version=str(self.version1.version))]
        self.mac_app = dict(appid=self.app.id, version=str(self.mac_version.short_version))

        self._generate_fake_statistics()
        now = datetime.now()
        updates = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), x - 1) for x in range(2, 13)]
        updates.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 0))
        installs = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), 1) for x in range(2, 13)]
        installs.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
        platform_statistics = dict(new=installs, updates=updates)
        self.users_statistics = dict(win=platform_statistics, mac=platform_statistics)
        self.data = dict(data=dict(self.users_statistics))

    @is_private()
    def test_unauthorized(self):
        client = APIClient()
        response = client.get(reverse(self.url, args=self.url_args), format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    @is_private()
    def test_list(self):
        response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(self.data).data, response.data)

    @freeze_time("2016-01-27")
    @is_private()
    def test_default_list(self):
        data_detail = self.data.copy()
        response = self.client.get(reverse(self.url, args=self.url_args), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(data_detail).data, response.data)


class StatisticsMonthsDetailTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-months-detail'
    url_args = ('app',)
    serializer = StatisticsMonthsSerializer
    url_get_params = dict(start='2015-2', end='2016-1')

    @is_private()
    def test_list(self):
        data_detail = self.data.copy()
        response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(data_detail).data, response.data)


class StatisticsVersionsTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-versions'
    url_args = ('app',)
    serializer = StatisticsMonthsSerializer
    url_get_params = dict(date='2016-01')

    @freeze_time("2016-01-27")
    def setUp(self):
        super(StatisticsVersionsTest, self).setUp()
        data = get_users_versions(self.app.id)
        self.data = dict(data=dict(data))


class StatisticsChannelsTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-channels'
    url_args = ('app',)
    serializer = StatisticsMonthsSerializer
    url_get_params = dict(date='2016-01')

    @freeze_time("2016-01-27")
    def setUp(self):
        super(StatisticsChannelsTest, self).setUp()
        data = get_channel_statistics(self.app.id)
        self.data = dict(data=dict(data))

    @is_private()
    def test_list(self):
        data_detail = self.data.copy()
        response = self.client.get(reverse(self.url, args=self.url_args), dict(date='2016-1'), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(data_detail).data, response.data)


class ServerVersionTest(APITestCase):
    url = 'api-version'
    serializer = ServerVersionSerializer

    def setUp(self):
        self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
        self.client.credentials(
            HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
        self.data = dict(version=settings.APP_VERSION)

    @is_private()
    def test(self):
        response = self.client.get(reverse(self.url), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(self.data).data, response.data)


1			# coding: utf8
2
3			"""
4			This software is licensed under the Apache 2 license, quoted below.
5
6			Copyright 2014 Crystalnix Limited
7
8			Licensed under the Apache License, Version 2.0 (the "License"); you may not
9			use this file except in compliance with the License. You may obtain a copy of
10			the License at
11
12			http://www.apache.org/licenses/LICENSE-2.0
13
14			Unless required by applicable law or agreed to in writing, software
15			distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16			WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17			License for the specific language governing permissions and limitations under
18			the License.
19			"""
20			from __future__ import unicode_literals
21			from builtins import bytes, range
22
23			import base64
24			from datetime import datetime, timedelta
25			from uuid import UUID
26
27			from django.core.urlresolvers import reverse
28			from django.contrib.auth import get_user_model
29			from django.core.files.uploadedfile import SimpleUploadedFile
30			from django.conf import settings
31
32			from rest_framework import status
33			from rest_framework.test import APITestCase, APIClient
34			from bitmapist import mark_event
35			from freezegun import freeze_time
36			import pytz
37
38			from omaha_server.utils import is_private
39			from omaha.statistics import userid_counting, get_users_versions, get_channel_statistics
40			from omaha.utils import redis
41			from omaha.serializers import (
42			AppSerializer,
43			DataSerializer,
44			PlatformSerializer,
45			ChannelSerializer,
46			VersionSerializer,
47			ActionSerializer,
48			StatisticsMonthsSerializer,
49			ServerVersionSerializer,
50			)
51			from omaha.factories import ApplicationFactory, DataFactory, PlatformFactory, ChannelFactory, VersionFactory, ActionFactory
52			from omaha.models import Application, Data, Channel, Platform, Version, Action
53			from omaha.tests.utils import temporary_media_root
54			from sparkle.models import SparkleVersion
55
56			User = get_user_model()
57
58
59			class BaseTest(object):
60			url = None
61			url_args = ()
62			url_detail = None
63			factory = None
64			serializer = None
65			maxDiff = None
66			is_private = True
67
68			def _is_private(self):
69			if not self.is_private and not settings.IS_PRIVATE:
70			return True
71			elif self.is_private and settings.IS_PRIVATE:
72			return True
73			else:
74			return False
75
76			def setUp(self):
77			self.objects = self.factory.create_batch(10)
78			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
79			self.client.credentials(
80			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
81
82			def test_unauthorized(self):
83			if not self._is_private():
84			return
85			client = APIClient()
86			response = client.get(reverse(self.url, args=self.url_args), format='json')
87			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
88
89			def test_list(self):
90			if not self._is_private():
91			return
92			response = self.client.get(reverse(self.url, args=self.url_args), format='json')
93			self.assertEqual(response.status_code, status.HTTP_200_OK)
94			self.assertEqual(len(response.data), 10)
95			self.assertEqual(self.serializer(self.objects, many=True).data, response.data[::-1])
96
97			def test_detail(self):
98			if not self._is_private():
99			return
100			obj = self.objects[0]
101			url = reverse(self.url_detail, kwargs=dict(pk=obj.pk))
102
103			response = self.client.get(url, format='json')
104			self.assertEqual(response.status_code, status.HTTP_200_OK)
105			self.assertDictEqual(response.data, self.serializer(obj).data)
106
107
108			class AppTest(BaseTest, APITestCase):
109			url = 'application-list'
110			url_detail = 'application-detail'
111			factory = ApplicationFactory
112			serializer = AppSerializer
113
114			@is_private()
115			def test_create(self):
116			data = dict(id='test_id', name='test_name', data_set=[])
117			response = self.client.post(reverse(self.url
118			), data, format='json')
119			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
120			obj = Application.objects.get(id=response.data['id'])
121			self.assertEqual(response.data, self.serializer(obj).data)
122
123
124			class DataTest(BaseTest, APITestCase):
			0 ignored issues – show Duplication introduced 2015-12-10 12:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
125			url = 'data-list'
126			url_detail = 'data-detail'
127			factory = DataFactory
128			serializer = DataSerializer
129
130			@is_private()
131			def test_create(self):
132			app = ApplicationFactory.create()
133			data = dict(name=0, app=app.pk)
134			response = self.client.post(reverse(self.url), data, format='json')
135			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
136			obj = Data.objects.get(id=response.data['id'])
137			self.assertEqual(response.data, self.serializer(obj).data)
138
139
140			class PlatformTest(BaseTest, APITestCase):
141			url = 'platform-list'
142			url_detail = 'platform-detail'
143			factory = PlatformFactory
144			serializer = PlatformSerializer
145
146			@is_private()
147			def test_list(self):
148			super(PlatformTest, self).test_list()
149
150			@is_private()
151			def test_create(self):
152			data = dict(name='test_name')
153			response = self.client.post(reverse(self.url), data, format='json')
154			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
155			obj = Platform.objects.get(id=response.data['id'])
156			self.assertEqual(response.data, self.serializer(obj).data)
157
158
159			class ChannelTest(BaseTest, APITestCase):
160			url = 'channel-list'
161			url_detail = 'channel-detail'
162			factory = ChannelFactory
163			serializer = ChannelSerializer
164
165			@is_private()
166			def test_create(self):
167			data = dict(name='test_name')
168			response = self.client.post(reverse(self.url), data, format='json')
169			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
170			obj = Channel.objects.get(id=response.data['id'])
171			self.assertEqual(response.data, self.serializer(obj).data)
172
173
174			class VersionTest(BaseTest, APITestCase):
175			url = 'version-list'
176			url_detail = 'version-detail'
177			factory = VersionFactory
178			serializer = VersionSerializer
179
180			@is_private()
181			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
182			def test_detail(self):
183			super(VersionTest, self).test_detail()
184
185			@is_private()
186			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
187			def test_list(self):
188			super(VersionTest, self).test_list()
189
190			@is_private()
191			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
192			def test_create(self):
193			data = dict(
194			app=ApplicationFactory.create().id,
195			platform=PlatformFactory.create().id,
196			channel=ChannelFactory.create().id,
197			version='1.2.3.4',
198			file=SimpleUploadedFile("chrome.exe", b'content'),
199			)
200			response = self.client.post(reverse(self.url), data)
201			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
202			version = Version.objects.get(id=response.data['id'])
203			self.assertEqual(response.data, self.serializer(version).data)
204			self.assertEqual(version.file_size, len(b'content'))
205			self.assertTrue(version.is_enabled)
206
207
208			class ActionTest(BaseTest, APITestCase):
			0 ignored issues – show Duplication introduced 2015-12-10 12:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
209			url = 'action-list'
210			url_detail = 'action-detail'
211			factory = ActionFactory
212			serializer = ActionSerializer
213
214			@is_private()
215			def test_create(self):
216			version = VersionFactory.create()
217			data = dict(event=1, version=version.pk)
218			response = self.client.post(reverse(self.url), data, format='json')
219			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
220			obj = Action.objects.get(id=response.data['id'])
221			self.assertEqual(response.data, self.serializer(obj).data)
222
223			class LiveStatistics(APITestCase):
224			maxDiff = None
225
226			def _generate_fake_statistics(self):
227			# now = datetime(2016, 2, 13)
228			date = datetime(2016, 2, 13, 0)
229			for i in range(self.n_hours):
230			for id in range(0, i):
231			mark_event('online:app:win:2.0.0.0', id, now=date, track_hourly=True)
232			mark_event('online:app:mac:4.0.0.1', id, now=date, track_hourly=True)
233			for id in range(i, self.n_hours):
234			mark_event('online:app:win:1.0.0.0', id, now=date, track_hourly=True)
235			mark_event('online:app:mac:3.0.0.0', id, now=date, track_hourly=True)
236			date += timedelta(hours=1)
237
238			def setUp(self):
239			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
240			self.client.credentials(
241			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
242
243			redis.flushdb()
244			self.app = Application.objects.create(id='app', name='app')
245			self.channel = Channel.objects.create(name='stable')
246			self.platform = Platform.objects.create(name='win')
247			self.version1 = Version.objects.create(
248			app=self.app,
249			platform=self.platform,
250			channel=self.channel,
251			version='1.0.0.0',
252			file=SimpleUploadedFile('./chrome_installer.exe', False))
253
254			self.version2 = Version.objects.create(
255			app=self.app,
256			platform=self.platform,
257			channel=self.channel,
258			version='2.0.0.0',
259			file=SimpleUploadedFile('./chrome_installer.exe', False))
260
261			self.sparkle_version1 = SparkleVersion.objects.create(
262			app=self.app,
263			channel=self.channel,
264			version='0.0',
265			short_version='3.0.0.0',
266			file=SimpleUploadedFile('./chrome_installer.dmg', False))
267
268			self.sparkle_version2 = SparkleVersion.objects.create(
269			app=self.app,
270			channel=self.channel,
271			version='0.1',
272			short_version='4.0.0.1',
273			file=SimpleUploadedFile('./chrome_installer.dmg', False))
274
275			self.n_hours = 36
276			self._generate_fake_statistics()
277
278			hours = [datetime(2016, 2, 13, 0, tzinfo=pytz.UTC) + timedelta(hours=hour)
279			for hour in range(self.n_hours)]
280			self.win_statistics = [('1.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
281			for (i, hour)in enumerate(hours)])]
282			self.win_statistics.append(('2.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
283			for (i, hour)in enumerate(hours)]))
284
285			self.mac_statistics = [('3.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
286			for (i, hour)in enumerate(hours)])]
287			self.mac_statistics.append(('4.0.0.1', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
288			for (i, hour)in enumerate(hours)]))
289
290			self.data = dict(data=dict(win=dict(self.win_statistics),
291			mac=dict(self.mac_statistics)))
292
293			@is_private()
294			def test_unauthorized(self):
295			client = APIClient()
296			response = client.get(reverse('api-statistics-live', args=('app',)), format='json')
297			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
298
299			@is_private()
300			def test_list(self):
301			start = datetime(2016, 2, 13, 0)
302			end = start + timedelta(hours=self.n_hours-1)
303			response = self.client.get(reverse('api-statistics-live', args=('app',)),
304			dict(start=start.isoformat(), end=end.isoformat()),
305			format='json')
306			self.assertEqual(response.status_code, status.HTTP_200_OK)
307			self.assertDictEqual(StatisticsMonthsSerializer(self.data).data, response.data)
308
309
310
311			class StatisticsMonthsMixin(object):
312			url = None
313			url_args = ()
314			serializer = None
315			url_get_params = None
316
317			def _generate_fake_statistics(self):
318			now = datetime.now()
319			prev_year = now.year - 1
320
321			for i in range(1, 13):
322			date = datetime(year=prev_year, month=i, day=10)
323			for id in range(1, i + 1):
324			user_id = UUID(int=id)
325			userid_counting(user_id, self.app_list, self.platform.name, now=date)
326			user_id = UUID(int=1000 + id)
327			userid_counting(user_id, [self.mac_app], 'mac', now=date)
328
329			user_id = UUID(int=13)
330			userid_counting(user_id, self.app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
331			user_id = UUID(int=1013)
332			userid_counting(user_id, [self.mac_app], 'mac', now=datetime(year=now.year, month=1, day=1))
333
334			@freeze_time("2016-01-27")
335			@temporary_media_root()
336			def setUp(self):
337			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
338			self.client.credentials(
339			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
340
341			redis.flushdb()
342			self.app = Application.objects.create(id='app', name='app')
343			self.channel = Channel.objects.create(name='stable')
344			self.platform = Platform.objects.create(name='win')
345			self.version1 = Version.objects.create(
346			app=self.app,
347			platform=self.platform,
348			channel=self.channel,
349			version='1.0.0.0',
350			file=SimpleUploadedFile('./chrome_installer.exe', False))
351			self.version2 = Version.objects.create(
352			app=self.app,
353			platform=self.platform,
354			channel=self.channel,
355			version='2.0.0.0',
356			file=SimpleUploadedFile('./chrome_installer.exe', False))
357			self.mac_version = SparkleVersion.objects.create(
358			app=self.app,
359			channel=self.channel,
360			version='782.112',
361			short_version='13.0.782.112',
362			dsa_signature='MCwCFCdoW13VBGJWIfIklKxQVyetgxE7AhQTVuY9uQT0KOV1UEk21epBsGZMPg==',
363			file=SimpleUploadedFile('./chrome.dmg', b'_' * 1024),
364			file_size=1024)
365			self.app_list = [dict(appid=self.app.id, version=str(self.version1.version))]
366			self.mac_app = dict(appid=self.app.id, version=str(self.mac_version.short_version))
367
368			self._generate_fake_statistics()
369			now = datetime.now()
370			updates = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), x - 1) for x in range(2, 13)]
371			updates.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 0))
372			installs = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), 1) for x in range(2, 13)]
373			installs.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
374			platform_statistics = dict(new=installs, updates=updates)
375			self.users_statistics = dict(win=platform_statistics, mac=platform_statistics)
376			self.data = dict(data=dict(self.users_statistics))
377
378			@is_private()
379			def test_unauthorized(self):
380			client = APIClient()
381			response = client.get(reverse(self.url, args=self.url_args), format='json')
382			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
383
384			@is_private()
385			def test_list(self):
386			response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
387			self.assertEqual(response.status_code, status.HTTP_200_OK)
388			self.assertEqual(self.serializer(self.data).data, response.data)
389
390			@freeze_time("2016-01-27")
391			@is_private()
392			def test_default_list(self):
393			data_detail = self.data.copy()
394			response = self.client.get(reverse(self.url, args=self.url_args), format='json')
395			self.assertEqual(response.status_code, status.HTTP_200_OK)
396			self.assertEqual(self.serializer(data_detail).data, response.data)
397
398
399			class StatisticsMonthsDetailTest(StatisticsMonthsMixin, APITestCase):
400			url = 'api-statistics-months-detail'
401			url_args = ('app',)
402			serializer = StatisticsMonthsSerializer
403			url_get_params = dict(start='2015-2', end='2016-1')
404
405			@is_private()
406			def test_list(self):
407			data_detail = self.data.copy()
408			response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
409			self.assertEqual(response.status_code, status.HTTP_200_OK)
410			self.assertEqual(self.serializer(data_detail).data, response.data)
411
412
413			class StatisticsVersionsTest(StatisticsMonthsMixin, APITestCase):
414			url = 'api-statistics-versions'
415			url_args = ('app',)
416			serializer = StatisticsMonthsSerializer
417			url_get_params = dict(date='2016-01')
418
419			@freeze_time("2016-01-27")
420			def setUp(self):
421			super(StatisticsVersionsTest, self).setUp()
422			data = get_users_versions(self.app.id)
423			self.data = dict(data=dict(data))
424
425
426			class StatisticsChannelsTest(StatisticsMonthsMixin, APITestCase):
427			url = 'api-statistics-channels'
428			url_args = ('app',)
429			serializer = StatisticsMonthsSerializer
430			url_get_params = dict(date='2016-01')
431
432			@freeze_time("2016-01-27")
433			def setUp(self):
434			super(StatisticsChannelsTest, self).setUp()
435			data = get_channel_statistics(self.app.id)
436			self.data = dict(data=dict(data))
437
438			@is_private()
439			def test_list(self):
440			data_detail = self.data.copy()
441			response = self.client.get(reverse(self.url, args=self.url_args), dict(date='2016-1'), format='json')
442			self.assertEqual(response.status_code, status.HTTP_200_OK)
443			self.assertEqual(self.serializer(data_detail).data, response.data)
444
445
446			class ServerVersionTest(APITestCase):
447			url = 'api-version'
448			serializer = ServerVersionSerializer
449
450			def setUp(self):
451			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
452			self.client.credentials(
453			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
454			self.data = dict(version=settings.APP_VERSION)
455
456			@is_private()
457			def test(self):
458			response = self.client.get(reverse(self.url), format='json')
459			self.assertEqual(response.status_code, status.HTTP_200_OK)
460			self.assertEqual(self.serializer(self.data).data, response.data)
461

Crystalnix / omaha-server

Push — master ( 67eae9...f0f1ec )

StatisticsChannelsTest.test_list() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like