VersionTest.test_update() - Code Metrics - Inspection of "Merge branch 'master' into dev" - Crystalnix/omaha-server - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 61d892...ba8a83 )

unknown

created 2016-10-04 10:30 UTC

VersionTest.test_update() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
dl	0
loc	24
rs	8.9713
c	1
b	0
f	0

# 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 lxml.builder import E
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 import fixtures
from omaha.tests.utils import temporary_media_root, create_app_xml
from sparkle.models import SparkleVersion
from sparkle.statistics import userid_counting as mac_userid_counting
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_403_FORBIDDEN)

    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)

    @is_private()
    @temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
    def test_update(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'),
            is_enabled=False,
        )
        response = self.client.post(reverse(self.url), data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        obj_id = response.data['id']
        version = Version.objects.get(id=obj_id)
        self.assertEqual(response.data, self.serializer(version).data)
        self.assertEqual(version.file_size, len(b'content'))
        self.assertFalse(version.is_enabled)
        url = reverse(self.url_detail, kwargs=dict(pk=obj_id))
        response = self.client.patch(url, dict(is_enabled=True))
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        version = Version.objects.get(id=obj_id)
        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_403_FORBIDDEN)

    @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.install_app_list, self.platform.name, now=date)
                user_id = UUID(int=1000 + id)
                mac_userid_counting(user_id, self.mac_app, 'mac', now=date)
            userid_counting(UUID(int=i), self.uninstall_app_list, self.platform.name, now=date)

        user_id = UUID(int=13)
        userid_counting(user_id, self.install_app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
        userid_counting(user_id, self.uninstall_app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
        user_id = UUID(int=1013)
        mac_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)
        app_kwargs = dict(appid=self.app.id, version=str(self.version1.version))
        install_app = create_app_xml(events=[fixtures.event_install_success], **app_kwargs)
        uninstall_app = create_app_xml(events=[fixtures.event_uninstall_success], **app_kwargs)
        self.install_app_list = [install_app]
        self.uninstall_app_list = [uninstall_app]
        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))
        uninstalls = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), 1) for x in range(2, 13)]
        uninstalls.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))

        win_platform_statistics = dict(new=installs, updates=updates, uninstalls=uninstalls)
        mac_platform_statistics = dict(new=installs, updates=updates)
        self.users_statistics = dict(win=win_platform_statistics, mac=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_403_FORBIDDEN)

    @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 lxml.builder import E
33		from rest_framework import status
34		from rest_framework.test import APITestCase, APIClient
35		from bitmapist import mark_event
36		from freezegun import freeze_time
37		import pytz
38
39		from omaha_server.utils import is_private
40		from omaha.statistics import userid_counting, get_users_versions, get_channel_statistics
41		from omaha.utils import redis
42		from omaha.serializers import (
43		AppSerializer,
44		DataSerializer,
45		PlatformSerializer,
46		ChannelSerializer,
47		VersionSerializer,
48		ActionSerializer,
49		StatisticsMonthsSerializer,
50		ServerVersionSerializer,
51		)
52		from omaha.factories import ApplicationFactory, DataFactory, PlatformFactory, ChannelFactory, VersionFactory, ActionFactory
53		from omaha.models import Application, Data, Channel, Platform, Version, Action
54		from omaha.tests import fixtures
55		from omaha.tests.utils import temporary_media_root, create_app_xml
56		from sparkle.models import SparkleVersion
57		from sparkle.statistics import userid_counting as mac_userid_counting
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_403_FORBIDDEN)
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	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-28 08:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-28 08:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
125
126		class DataTest(BaseTest, APITestCase):
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	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-28 08:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-28 08:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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		@is_private()
210		@temporary_media_root(MEDIA_URL='http://cache.pack.google.com/edgedl/chrome/install/782.112/')
211		def test_update(self):
212		data = dict(
213		app=ApplicationFactory.create().id,
214		platform=PlatformFactory.create().id,
215		channel=ChannelFactory.create().id,
216		version='1.2.3.4',
217		file=SimpleUploadedFile("chrome.exe", b'content'),
218		is_enabled=False,
219		)
220		response = self.client.post(reverse(self.url), data)
221		self.assertEqual(response.status_code, status.HTTP_201_CREATED)
222		obj_id = response.data['id']
223		version = Version.objects.get(id=obj_id)
224		self.assertEqual(response.data, self.serializer(version).data)
225		self.assertEqual(version.file_size, len(b'content'))
226		self.assertFalse(version.is_enabled)
227		url = reverse(self.url_detail, kwargs=dict(pk=obj_id))
228		response = self.client.patch(url, dict(is_enabled=True))
229		self.assertEqual(response.status_code, status.HTTP_200_OK)
230		version = Version.objects.get(id=obj_id)
231		self.assertEqual(version.file_size, len(b'content'))
232		self.assertTrue(version.is_enabled)
233
234
235		class ActionTest(BaseTest, APITestCase):
236		url = 'action-list'
237		url_detail = 'action-detail'
238		factory = ActionFactory
239		serializer = ActionSerializer
240
241		@is_private()
242		def test_create(self):
243		version = VersionFactory.create()
244		data = dict(event=1, version=version.pk)
245		response = self.client.post(reverse(self.url), data, format='json')
246		self.assertEqual(response.status_code, status.HTTP_201_CREATED)
247		obj = Action.objects.get(id=response.data['id'])
248		self.assertEqual(response.data, self.serializer(obj).data)
249
250		class LiveStatistics(APITestCase):
251		maxDiff = None
252
253		def _generate_fake_statistics(self):
254		# now = datetime(2016, 2, 13)
255		date = datetime(2016, 2, 13, 0)
256		for i in range(self.n_hours):
257		for id in range(0, i):
258		mark_event('online:app:win:2.0.0.0', id, now=date, track_hourly=True)
259		mark_event('online:app:mac:4.0.0.1', id, now=date, track_hourly=True)
260		for id in range(i, self.n_hours):
261		mark_event('online:app:win:1.0.0.0', id, now=date, track_hourly=True)
262		mark_event('online:app:mac:3.0.0.0', id, now=date, track_hourly=True)
263		date += timedelta(hours=1)
264
265		def setUp(self):
266		self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
267		self.client.credentials(
268		HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
269
270		redis.flushdb()
271		self.app = Application.objects.create(id='app', name='app')
272		self.channel = Channel.objects.create(name='stable')
273		self.platform = Platform.objects.create(name='win')
274		self.version1 = Version.objects.create(
275		app=self.app,
276		platform=self.platform,
277		channel=self.channel,
278		version='1.0.0.0',
279		file=SimpleUploadedFile('./chrome_installer.exe', False))
280
281		self.version2 = Version.objects.create(
282		app=self.app,
283		platform=self.platform,
284		channel=self.channel,
285		version='2.0.0.0',
286		file=SimpleUploadedFile('./chrome_installer.exe', False))
287
288		self.sparkle_version1 = SparkleVersion.objects.create(
289		app=self.app,
290		channel=self.channel,
291		version='0.0',
292		short_version='3.0.0.0',
293		file=SimpleUploadedFile('./chrome_installer.dmg', False))
294
295		self.sparkle_version2 = SparkleVersion.objects.create(
296		app=self.app,
297		channel=self.channel,
298		version='0.1',
299		short_version='4.0.0.1',
300		file=SimpleUploadedFile('./chrome_installer.dmg', False))
301
302		self.n_hours = 36
303		self._generate_fake_statistics()
304
305		hours = [datetime(2016, 2, 13, 0, tzinfo=pytz.UTC) + timedelta(hours=hour)
306		for hour in range(self.n_hours)]
307		self.win_statistics = [('1.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
308		for (i, hour)in enumerate(hours)])]
309		self.win_statistics.append(('2.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
310		for (i, hour)in enumerate(hours)]))
311
312		self.mac_statistics = [('3.0.0.0', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.n_hours - i]
313		for (i, hour)in enumerate(hours)])]
314		self.mac_statistics.append(('4.0.0.1', [[hour.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), i]
315		for (i, hour)in enumerate(hours)]))
316
317		self.data = dict(data=dict(win=dict(self.win_statistics),
318		mac=dict(self.mac_statistics)))
319
320		@is_private()
321		def test_unauthorized(self):
322		client = APIClient()
323		response = client.get(reverse('api-statistics-live', args=('app',)), format='json')
324		self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
325
326		@is_private()
327		def test_list(self):
328		start = datetime(2016, 2, 13, 0)
329		end = start + timedelta(hours=self.n_hours-1)
330		response = self.client.get(reverse('api-statistics-live', args=('app',)),
331		dict(start=start.isoformat(), end=end.isoformat()),
332		format='json')
333		self.assertEqual(response.status_code, status.HTTP_200_OK)
334		self.assertDictEqual(StatisticsMonthsSerializer(self.data).data, response.data)
335
336
337		class StatisticsMonthsMixin(object):
338		url = None
339		url_args = ()
340		serializer = None
341		url_get_params = None
342
343		def _generate_fake_statistics(self):
344		now = datetime.now()
345		prev_year = now.year - 1
346
347		for i in range(1, 13):
348		date = datetime(year=prev_year, month=i, day=10)
349		for id in range(1, i + 1):
350		user_id = UUID(int=id)
351		userid_counting(user_id, self.install_app_list, self.platform.name, now=date)
352		user_id = UUID(int=1000 + id)
353		mac_userid_counting(user_id, self.mac_app, 'mac', now=date)
354		userid_counting(UUID(int=i), self.uninstall_app_list, self.platform.name, now=date)
355
356		user_id = UUID(int=13)
357		userid_counting(user_id, self.install_app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
358		userid_counting(user_id, self.uninstall_app_list, self.platform.name, now=datetime(year=now.year, month=1, day=1))
359		user_id = UUID(int=1013)
360		mac_userid_counting(user_id, self.mac_app, 'mac', now=datetime(year=now.year, month=1, day=1))
361
362		@freeze_time("2016-01-27")
363		@temporary_media_root()
364		def setUp(self):
365		self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
366		self.client.credentials(
367		HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
368
369		redis.flushdb()
370		self.app = Application.objects.create(id='app', name='app')
371		self.channel = Channel.objects.create(name='stable')
372		self.platform = Platform.objects.create(name='win')
373		self.version1 = Version.objects.create(
374		app=self.app,
375		platform=self.platform,
376		channel=self.channel,
377		version='1.0.0.0',
378		file=SimpleUploadedFile('./chrome_installer.exe', False))
379		self.version2 = Version.objects.create(
380		app=self.app,
381		platform=self.platform,
382		channel=self.channel,
383		version='2.0.0.0',
384		file=SimpleUploadedFile('./chrome_installer.exe', False))
385		self.mac_version = SparkleVersion.objects.create(
386		app=self.app,
387		channel=self.channel,
388		version='782.112',
389		short_version='13.0.782.112',
390		dsa_signature='MCwCFCdoW13VBGJWIfIklKxQVyetgxE7AhQTVuY9uQT0KOV1UEk21epBsGZMPg==',
391		file=SimpleUploadedFile('./chrome.dmg', b'_' * 1024),
392		file_size=1024)
393		app_kwargs = dict(appid=self.app.id, version=str(self.version1.version))
394		install_app = create_app_xml(events=[fixtures.event_install_success], **app_kwargs)
395		uninstall_app = create_app_xml(events=[fixtures.event_uninstall_success], **app_kwargs)
396		self.install_app_list = [install_app]
397		self.uninstall_app_list = [uninstall_app]
398		self.mac_app = dict(appid=self.app.id, version=str(self.mac_version.short_version))
399	View Code Duplication
		0 ignored issues – show Duplication introduced 2016-04-28 08:43 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
400		self._generate_fake_statistics()
401		now = datetime.now()
402		updates = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), x - 1) for x in range(2, 13)]
403		updates.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 0))
404		installs = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), 1) for x in range(2, 13)]
405		installs.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
406		uninstalls = [(datetime(now.year-1, x, 1).strftime("%Y-%m"), 1) for x in range(2, 13)]
407		uninstalls.append((datetime(now.year, 1, 1).strftime("%Y-%m"), 1))
408
409		win_platform_statistics = dict(new=installs, updates=updates, uninstalls=uninstalls)
410		mac_platform_statistics = dict(new=installs, updates=updates)
411		self.users_statistics = dict(win=win_platform_statistics, mac=mac_platform_statistics)
412		self.data = dict(data=dict(self.users_statistics))
413
414		@is_private()
415		def test_unauthorized(self):
416		client = APIClient()
417		response = client.get(reverse(self.url, args=self.url_args), format='json')
418		self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
419
420		@is_private()
421		def test_list(self):
422		response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
423		self.assertEqual(response.status_code, status.HTTP_200_OK)
424		self.assertEqual(self.serializer(self.data).data, response.data)
425
426		@freeze_time("2016-01-27")
427		@is_private()
428		def test_default_list(self):
429		data_detail = self.data.copy()
430		response = self.client.get(reverse(self.url, args=self.url_args), format='json')
431		self.assertEqual(response.status_code, status.HTTP_200_OK)
432		self.assertEqual(self.serializer(data_detail).data, response.data)
433
434
435		class StatisticsMonthsDetailTest(StatisticsMonthsMixin, APITestCase):
436		url = 'api-statistics-months-detail'
437		url_args = ('app',)
438		serializer = StatisticsMonthsSerializer
439		url_get_params = dict(start='2015-2', end='2016-1')
440
441		@is_private()
442		def test_list(self):
443		data_detail = self.data.copy()
444		response = self.client.get(reverse(self.url, args=self.url_args), self.url_get_params, format='json')
445		self.assertEqual(response.status_code, status.HTTP_200_OK)
446		self.assertEqual(self.serializer(data_detail).data, response.data)
447
448
449		class StatisticsVersionsTest(StatisticsMonthsMixin, APITestCase):
450		url = 'api-statistics-versions'
451		url_args = ('app',)
452		serializer = StatisticsMonthsSerializer
453		url_get_params = dict(date='2016-01')
454
455		@freeze_time("2016-01-27")
456		def setUp(self):
457		super(StatisticsVersionsTest, self).setUp()
458		data = get_users_versions(self.app.id)
459		self.data = dict(data=dict(data))
460
461
462		class StatisticsChannelsTest(StatisticsMonthsMixin, APITestCase):
463		url = 'api-statistics-channels'
464		url_args = ('app',)
465		serializer = StatisticsMonthsSerializer
466		url_get_params = dict(date='2016-01')
467
468		@freeze_time("2016-01-27")
469		def setUp(self):
470		super(StatisticsChannelsTest, self).setUp()
471		data = get_channel_statistics(self.app.id)
472		self.data = dict(data=dict(data))
473
474		@is_private()
475		def test_list(self):
476		data_detail = self.data.copy()
477		response = self.client.get(reverse(self.url, args=self.url_args), dict(date='2016-1'), format='json')
478		self.assertEqual(response.status_code, status.HTTP_200_OK)
479		self.assertEqual(self.serializer(data_detail).data, response.data)
480
481
482		class ServerVersionTest(APITestCase):
483		url = 'api-version'
484		serializer = ServerVersionSerializer
485
486		def setUp(self):
487		self.user = User.objects.create_user(username='test', password='secret', email='[email protected]')
488		self.client.credentials(
489		HTTP_AUTHORIZATION='Basic %s' % base64.b64encode(bytes('{}:{}'.format('test', 'secret'), 'utf8')).decode())
490		self.data = dict(version=settings.APP_VERSION)
491
492		@is_private()
493		def test(self):
494		response = self.client.get(reverse(self.url), format='json')
495		self.assertEqual(response.status_code, status.HTTP_200_OK)
496		self.assertEqual(self.serializer(self.data).data, response.data)
497

Crystalnix / omaha-server

Push — master ( 61d892...ba8a83 )

VersionTest.test_update() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like