LiveStatistics.test_unauthorized() - Code Metrics - Inspection of "Bump version: 0.2.5 → 0.2.6" - Crystalnix/omaha-server - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9ed73f...7648fa )

unknown

created 2016-02-19 13:15 UTC

LiveStatistics.test_unauthorized() A

↳ Parent: LiveStatistics

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	5
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
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 freezegun import freeze_time

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)

        for i in range(self.n_hours):
            date = datetime(now.year, now.month, now.day, i)
            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)


    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 = 24
        self._generate_fake_statistics()

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

        self.mac_statistics = [('3.0.0.0', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - hour]
                                            for hour in range(self.n_hours)])]
        self.mac_statistics.append(('4.0.0.1', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), hour]
                                                for hour in range(self.n_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):
        with freeze_time("2016-02-13 23:00:01"):
            response = self.client.get(reverse('api-statistics-live', args=('app',)), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertDictEqual(StatisticsMonthsSerializer(self.data).data, response.data)


@freeze_time("2016-01-27")
class StatisticsMonthsMixin(object):
    url = None
    url_args = ()
    serializer = None

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

        for i in range(2, 13):
            date = datetime(year=prev_year, month=i, day=1)
            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=13)
        userid_counting(user_id, self.app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))

    @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.app_list = [dict(appid=self.app.id, version=str(self.version1.version))]

        self._generate_fake_statistics()
        now = datetime.now()
        self.users_statistics = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), x) for x in range(2, 13)]
        self.users_statistics.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
        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), format='json')
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.serializer(self.data).data, response.data)


class StatisticsMonthsListTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-months-list'
    serializer = StatisticsMonthsSerializer

@freeze_time("2016-01-27")
class StatisticsMonthsDetailTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-months-detail'
    url_args = ('app',)
    serializer = StatisticsMonthsSerializer

    @is_private()
    def test_list(self):
        data_detail = self.data.copy()
        data_detail['data']['install_count'] = 0
        data_detail['data']['update_count'] = 0
        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 StatisticsVersionsTest(StatisticsMonthsMixin, APITestCase):
    url = 'api-statistics-versions'
    url_args = ('app',)
    serializer = StatisticsMonthsSerializer

    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

    def setUp(self):
        super(StatisticsChannelsTest, self).setUp()
        data = get_channel_statistics(self.app.id)
        self.data = dict(data=dict(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
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 freezegun import freeze_time
35
36			from bitmapist import mark_event
37			from freezegun import freeze_time
38			import pytz
39
40			from omaha_server.utils import is_private
41			from omaha.statistics import userid_counting, get_users_versions, get_channel_statistics
42			from omaha.utils import redis
43			from omaha.serializers import (
44			AppSerializer,
45			DataSerializer,
46			PlatformSerializer,
47			ChannelSerializer,
48			VersionSerializer,
49			ActionSerializer,
50			StatisticsMonthsSerializer,
51			ServerVersionSerializer,
52			)
53			from omaha.factories import ApplicationFactory, DataFactory, PlatformFactory, ChannelFactory, VersionFactory, ActionFactory
54			from omaha.models import Application, Data, Channel, Platform, Version, Action
55			from omaha.tests.utils import temporary_media_root
56			from sparkle.models import SparkleVersion
57
58			User = get_user_model()
59
60
61			class BaseTest(object):
62			url = None
63			url_args = ()
64			url_detail = None
65			factory = None
66			serializer = None
67			maxDiff = None
68			is_private = True
69
70			def _is_private(self):
71			if not self.is_private and not settings.IS_PRIVATE:
72			return True
73			elif self.is_private and settings.IS_PRIVATE:
74			return True
75			else:
76			return False
77
78			def setUp(self):
79			self.objects = self.factory.create_batch(10)
80			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
81			self.client.credentials(
82			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
83
84			def test_unauthorized(self):
85			if not self._is_private():
86			return
87			client = APIClient()
88			response = client.get(reverse(self.url, args=self.url_args), format='json')
89			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
90
91			def test_list(self):
92			if not self._is_private():
93			return
94			response = self.client.get(reverse(self.url, args=self.url_args), format='json')
95			self.assertEqual(response.status_code, status.HTTP_200_OK)
96			self.assertEqual(len(response.data), 10)
97			self.assertEqual(self.serializer(self.objects, many=True).data, response.data[::-1])
98
99			def test_detail(self):
100			if not self._is_private():
101			return
102			obj = self.objects[0]
103			url = reverse(self.url_detail, kwargs=dict(pk=obj.pk))
104
105			response = self.client.get(url, format='json')
106			self.assertEqual(response.status_code, status.HTTP_200_OK)
107			self.assertDictEqual(response.data, self.serializer(obj).data)
108
109
110			class AppTest(BaseTest, APITestCase):
111			url = 'application-list'
112			url_detail = 'application-detail'
113			factory = ApplicationFactory
114			serializer = AppSerializer
115
116			@is_private()
117			def test_create(self):
118			data = dict(id='test_id', name='test_name', data_set=[])
119			response = self.client.post(reverse(self.url
120			), data, format='json')
121			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
122			obj = Application.objects.get(id=response.data['id'])
123			self.assertEqual(response.data, self.serializer(obj).data)
124
125
126			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...
127			url = 'data-list'
128			url_detail = 'data-detail'
129			factory = DataFactory
130			serializer = DataSerializer
131
132			@is_private()
133			def test_create(self):
134			app = ApplicationFactory.create()
135			data = dict(name=0, app=app.pk)
136			response = self.client.post(reverse(self.url), data, format='json')
137			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
138			obj = Data.objects.get(id=response.data['id'])
139			self.assertEqual(response.data, self.serializer(obj).data)
140
141
142			class PlatformTest(BaseTest, APITestCase):
143			url = 'platform-list'
144			url_detail = 'platform-detail'
145			factory = PlatformFactory
146			serializer = PlatformSerializer
147
148			@is_private()
149			def test_list(self):
150			super(PlatformTest, self).test_list()
151
152			@is_private()
153			def test_create(self):
154			data = dict(name='test_name')
155			response = self.client.post(reverse(self.url), data, format='json')
156			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
157			obj = Platform.objects.get(id=response.data['id'])
158			self.assertEqual(response.data, self.serializer(obj).data)
159
160
161			class ChannelTest(BaseTest, APITestCase):
162			url = 'channel-list'
163			url_detail = 'channel-detail'
164			factory = ChannelFactory
165			serializer = ChannelSerializer
166
167			@is_private()
168			def test_create(self):
169			data = dict(name='test_name')
170			response = self.client.post(reverse(self.url), data, format='json')
171			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
172			obj = Channel.objects.get(id=response.data['id'])
173			self.assertEqual(response.data, self.serializer(obj).data)
174
175
176			class VersionTest(BaseTest, APITestCase):
177			url = 'version-list'
178			url_detail = 'version-detail'
179			factory = VersionFactory
180			serializer = VersionSerializer
181
182			@is_private()
183			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
184			def test_detail(self):
185			super(VersionTest, self).test_detail()
186
187			@is_private()
188			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
189			def test_list(self):
190			super(VersionTest, self).test_list()
191
192			@is_private()
193			@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
194			def test_create(self):
195			data = dict(
196			app=ApplicationFactory.create().id,
197			platform=PlatformFactory.create().id,
198			channel=ChannelFactory.create().id,
199			version='1.2.3.4',
200			file=SimpleUploadedFile("chrome.exe", b'content'),
201			)
202			response = self.client.post(reverse(self.url), data)
203			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
204			version = Version.objects.get(id=response.data['id'])
205			self.assertEqual(response.data, self.serializer(version).data)
206			self.assertEqual(version.file_size, len(b'content'))
207			self.assertTrue(version.is_enabled)
208
209
210			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...
211			url = 'action-list'
212			url_detail = 'action-detail'
213			factory = ActionFactory
214			serializer = ActionSerializer
215
216			@is_private()
217			def test_create(self):
218			version = VersionFactory.create()
219			data = dict(event=1, version=version.pk)
220			response = self.client.post(reverse(self.url), data, format='json')
221			self.assertEqual(response.status_code, status.HTTP_201_CREATED)
222			obj = Action.objects.get(id=response.data['id'])
223			self.assertEqual(response.data, self.serializer(obj).data)
224
225			class LiveStatistics(APITestCase):
226			maxDiff = None
227
228			def _generate_fake_statistics(self):
229			now = datetime(2016, 2, 13)
230
231			for i in range(self.n_hours):
232			date = datetime(now.year, now.month, now.day, i)
233			for id in range(0, i):
234			mark_event('online:app:win:2.0.0.0', id, now=date, track_hourly=True)
235
236			mark_event('online:app:mac:4.0.0.1', id, now=date, track_hourly=True)
237			for id in range(i, self.n_hours):
238			mark_event('online:app:win:1.0.0.0', id, now=date, track_hourly=True)
239			mark_event('online:app:mac:3.0.0.0', id, now=date, track_hourly=True)
240
241
242			def setUp(self):
243			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
244			self.client.credentials(
245			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
246
247			redis.flushdb()
248			self.app = Application.objects.create(id='app', name='app')
249			self.channel = Channel.objects.create(name='stable')
250			self.platform = Platform.objects.create(name='win')
251			self.version1 = Version.objects.create(
252			app=self.app,
253			platform=self.platform,
254			channel=self.channel,
255			version='1.0.0.0',
256			file=SimpleUploadedFile('./chrome_installer.exe', False))
257
258			self.version2 = Version.objects.create(
259			app=self.app,
260			platform=self.platform,
261			channel=self.channel,
262			version='2.0.0.0',
263			file=SimpleUploadedFile('./chrome_installer.exe', False))
264
265			self.sparkle_version1 = SparkleVersion.objects.create(
266			app=self.app,
267			channel=self.channel,
268			version='0.0',
269			short_version='3.0.0.0',
270			file=SimpleUploadedFile('./chrome_installer.dmg', False))
271
272			self.sparkle_version2 = SparkleVersion.objects.create(
273			app=self.app,
274			channel=self.channel,
275			version='0.1',
276			short_version='4.0.0.1',
277			file=SimpleUploadedFile('./chrome_installer.dmg', False))
278
279			self.n_hours = 24
280			self._generate_fake_statistics()
281
282			self.win_statistics = [('1.0.0.0', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - hour]
283			for hour in range(self.n_hours)])]
284			self.win_statistics.append(('2.0.0.0', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), hour]
285			for hour in range(self.n_hours)]))
286
287			self.mac_statistics = [('3.0.0.0', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - hour]
288			for hour in range(self.n_hours)])]
289			self.mac_statistics.append(('4.0.0.1', [[datetime(2016, 2, 13, hour, tzinfo=pytz.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), hour]
290			for hour in range(self.n_hours)]))
291
292			self.data = dict(data=dict(win=dict(self.win_statistics),
293			mac=dict(self.mac_statistics)))
294
295			@is_private()
296			def test_unauthorized(self):
297			client = APIClient()
298			response = client.get(reverse('api-statistics-live', args=('app',)), format='json')
299			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
300
301			@is_private()
302			def test_list(self):
303			with freeze_time("2016-02-13 23:00:01"):
304			response = self.client.get(reverse('api-statistics-live', args=('app',)), format='json')
305			self.assertEqual(response.status_code, status.HTTP_200_OK)
306			self.assertDictEqual(StatisticsMonthsSerializer(self.data).data, response.data)
307
308
309			@freeze_time("2016-01-27")
310			class StatisticsMonthsMixin(object):
311			url = None
312			url_args = ()
313			serializer = None
314
315			def _generate_fake_statistics(self):
316			now = datetime.now()
317			prev_year = now.year - 1
318
319			for i in range(2, 13):
320			date = datetime(year=prev_year, month=i, day=1)
321			for id in range(1, i + 1):
322			user_id = UUID(int=id)
323			userid_counting(user_id, self.app_list, self.platform.name, now=date)
324
325			user_id = UUID(int=13)
326			userid_counting(user_id, self.app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
327
328			@temporary_media_root()
329			def setUp(self):
330			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
331			self.client.credentials(
332			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
333
334			redis.flushdb()
335			self.app = Application.objects.create(id='app', name='app')
336			self.channel = Channel.objects.create(name='stable')
337			self.platform = Platform.objects.create(name='win')
338			self.version1 = Version.objects.create(
339			app=self.app,
340			platform=self.platform,
341			channel=self.channel,
342			version='1.0.0.0',
343			file=SimpleUploadedFile('./chrome_installer.exe', False))
344			self.version2 = Version.objects.create(
345			app=self.app,
346			platform=self.platform,
347			channel=self.channel,
348			version='2.0.0.0',
349			file=SimpleUploadedFile('./chrome_installer.exe', False))
350			self.app_list = [dict(appid=self.app.id, version=str(self.version1.version))]
351
352			self._generate_fake_statistics()
353			now = datetime.now()
354			self.users_statistics = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), x) for x in range(2, 13)]
355			self.users_statistics.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
356			self.data = dict(data=dict(self.users_statistics))
357
358			@is_private()
359			def test_unauthorized(self):
360			client = APIClient()
361			response = client.get(reverse(self.url, args=self.url_args), format='json')
362			self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
363
364
365			@is_private()
366			def test_list(self):
367			response = self.client.get(reverse(self.url, args=self.url_args), format='json')
368			self.assertEqual(response.status_code, status.HTTP_200_OK)
369			self.assertEqual(self.serializer(self.data).data, response.data)
370
371
372			class StatisticsMonthsListTest(StatisticsMonthsMixin, APITestCase):
373			url = 'api-statistics-months-list'
374			serializer = StatisticsMonthsSerializer
375
376			@freeze_time("2016-01-27")
377			class StatisticsMonthsDetailTest(StatisticsMonthsMixin, APITestCase):
378			url = 'api-statistics-months-detail'
379			url_args = ('app',)
380			serializer = StatisticsMonthsSerializer
381
382			@is_private()
383			def test_list(self):
384			data_detail = self.data.copy()
385			data_detail['data']['install_count'] = 0
386			data_detail['data']['update_count'] = 0
387			response = self.client.get(reverse(self.url, args=self.url_args), format='json')
388			self.assertEqual(response.status_code, status.HTTP_200_OK)
389			self.assertEqual(self.serializer(data_detail).data, response.data)
390
391
392			class StatisticsVersionsTest(StatisticsMonthsMixin, APITestCase):
393			url = 'api-statistics-versions'
394			url_args = ('app',)
395			serializer = StatisticsMonthsSerializer
396
397			def setUp(self):
398			super(StatisticsVersionsTest, self).setUp()
399			data = get_users_versions(self.app.id)
400			self.data = dict(data=dict(data))
401
402
403			class StatisticsChannelsTest(StatisticsMonthsMixin, APITestCase):
404			url = 'api-statistics-channels'
405			url_args = ('app',)
406			serializer = StatisticsMonthsSerializer
407
408			def setUp(self):
409			super(StatisticsChannelsTest, self).setUp()
410			data = get_channel_statistics(self.app.id)
411			self.data = dict(data=dict(data))
412
413			class ServerVersionTest(APITestCase):
414			url = 'api-version'
415			serializer = ServerVersionSerializer
416
417			def setUp(self):
418			self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
419			self.client.credentials(
420			HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
421			self.data = dict(version=settings.APP_VERSION)
422
423			@is_private()
424			def test(self):
425			response = self.client.get(reverse(self.url), format='json')
426			self.assertEqual(response.status_code, status.HTTP_200_OK)
427			self.assertEqual(self.serializer(self.data).data, response.data)
428

Crystalnix / omaha-server

Push — master ( 9ed73f...7648fa )

LiveStatistics.test_unauthorized() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like