Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use aggregate method in yandex-tank

Best Python code snippet using yandex-tank

tests.py

Source:tests.py

...54                        ['<SetAggregateInfoStep: set_aggregate_info>',55                        '<AddHostsToAggregateStep: add_host_to_aggregate>'])56    @test.create_stubs({api.nova: ('host_list', 'aggregate_details_list',57                                   'aggregate_create'), })58    def test_create_aggregate(self):59        aggregate = self.aggregates.first()60        api.nova.host_list(IsA(http.HttpRequest)).AndReturn(self.hosts.list())61        api.nova.aggregate_details_list(IsA(http.HttpRequest)).AndReturn([])62        workflow_data = self._get_create_workflow_data(aggregate)63        api.nova.aggregate_create(IsA(http.HttpRequest),64                                  name=workflow_data['name'],65                                  availability_zone=66                                  workflow_data['availability_zone'])\67            .AndReturn(aggregate)68        self.mox.ReplayAll()69        url = reverse(constants.AGGREGATES_CREATE_URL)70        res = self.client.post(url, workflow_data)71        self.assertNoFormErrors(res)72        self.assertRedirectsNoFollow(res,73                                     reverse(constants.AGGREGATES_INDEX_URL))74    @test.create_stubs({api.nova: ('host_list',75                                   'aggregate_details_list',76                                   'aggregate_create',77                                   'add_host_to_aggregate'), })78    def test_create_aggregate_with_hosts(self):79        aggregate = self.aggregates.first()80        hosts = self.hosts.list()81        api.nova.host_list(IsA(http.HttpRequest)).AndReturn(self.hosts.list())82        api.nova.aggregate_details_list(IsA(http.HttpRequest)).AndReturn([])83        workflow_data = self._get_create_workflow_data(aggregate, hosts)84        api.nova.aggregate_create(IsA(http.HttpRequest),85                                  name=workflow_data['name'],86                                  availability_zone=87                                  workflow_data['availability_zone'])\88            .AndReturn(aggregate)89        compute_hosts = []90        for host in hosts:91            if host.service == 'compute':92                compute_hosts.append(host)93        for host in compute_hosts:94            api.nova.add_host_to_aggregate(IsA(http.HttpRequest),95                                           aggregate.id, host.host_name)96        self.mox.ReplayAll()97        url = reverse(constants.AGGREGATES_CREATE_URL)98        res = self.client.post(url, workflow_data)99        self.assertNoFormErrors(res)100        self.assertRedirectsNoFollow(res,101                                     reverse(constants.AGGREGATES_INDEX_URL))102    @test.create_stubs({api.nova: ('host_list', 'aggregate_details_list', ), })103    def test_host_list_nova_compute(self):104        hosts = self.hosts.list()105        compute_hosts = []106        for host in hosts:107            if host.service == 'compute':108                compute_hosts.append(host)109        api.nova.host_list(IsA(http.HttpRequest)).AndReturn(self.hosts.list())110        self.mox.ReplayAll()111        url = reverse(constants.AGGREGATES_CREATE_URL)112        res = self.client.get(url)113        workflow = res.context['workflow']114        step = workflow.get_step("add_host_to_aggregate")115        field_name = step.get_member_field_name('member')116        self.assertEqual(len(step.action.fields[field_name].choices),117                         len(compute_hosts))118class AggregatesViewTests(test.BaseAdminViewTests):119    @test.create_stubs({api.nova: ('aggregate_details_list',120                                   'availability_zone_list',), })121    def test_index(self):122        api.nova.aggregate_details_list(IsA(http.HttpRequest)) \123                .AndReturn(self.aggregates.list())124        api.nova.availability_zone_list(IsA(http.HttpRequest), detailed=True) \125                .AndReturn(self.availability_zones.list())126        self.mox.ReplayAll()127        res = self.client.get(reverse(constants.AGGREGATES_INDEX_URL))128        self.assertTemplateUsed(res, constants.AGGREGATES_INDEX_VIEW_TEMPLATE)129        self.assertItemsEqual(res.context['host_aggregates_table'].data,130                              self.aggregates.list())131        self.assertItemsEqual(res.context['availability_zones_table'].data,132                              self.availability_zones.list())133    @test.create_stubs({api.nova: ('aggregate_update', 'aggregate_get',), })134    def _test_generic_update_aggregate(self, form_data, aggregate,135                                       error_count=0,136                                       expected_error_message=None):137        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id))\138                .AndReturn(aggregate)139        if not expected_error_message:140            az = form_data['availability_zone']141            aggregate_data = {'name': form_data['name'],142                              'availability_zone': az}143            api.nova.aggregate_update(IsA(http.HttpRequest), str(aggregate.id),144                                      aggregate_data)145        self.mox.ReplayAll()146        res = self.client.post(reverse(constants.AGGREGATES_UPDATE_URL,147                               args=[aggregate.id]),148                               form_data)149        if not expected_error_message:150            self.assertNoFormErrors(res)151            self.assertRedirectsNoFollow(res,152                    reverse(constants.AGGREGATES_INDEX_URL))153        else:154            self.assertFormErrors(res, error_count, expected_error_message)155    def test_update_aggregate(self):156        aggregate = self.aggregates.first()157        form_data = {'id': aggregate.id,158                     'name': 'my_new_name',159                     'availability_zone': 'my_new_zone'}160        self._test_generic_update_aggregate(form_data, aggregate)161    def test_update_aggregate_fails_missing_fields(self):162        aggregate = self.aggregates.first()163        form_data = {'id': aggregate.id}164        self._test_generic_update_aggregate(form_data, aggregate, 1,165                                            u'This field is required')166class ManageHostsTests(test.BaseAdminViewTests):167    @test.create_stubs({api.nova: ('aggregate_get', 'host_list')})168    def test_manage_hosts(self):169        aggregate = self.aggregates.first()170        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \171                .AndReturn(aggregate)172        api.nova.host_list(IsA(http.HttpRequest)) \173                .AndReturn(self.hosts.list())174        self.mox.ReplayAll()175        res = self.client.get(reverse(constants.AGGREGATES_MANAGE_HOSTS_URL,176                                      args=[aggregate.id]))177        self.assertEqual(res.status_code, 200)178        self.assertTemplateUsed(res,179                                constants.AGGREGATES_MANAGE_HOSTS_TEMPLATE)180    @test.create_stubs({api.nova: ('aggregate_get', 'add_host_to_aggregate',181                                   'remove_host_from_aggregate',182                                   'host_list')})183    def test_manage_hosts_update_add_remove_not_empty_aggregate(self):184        aggregate = self.aggregates.first()185        aggregate.hosts = ['host1', 'host2']186        host = self.hosts.list()[0]187        form_data = {'manageaggregatehostsaction_role_member':188                     [host.host_name]}189        api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),190                                            str(aggregate.id),191                                            'host2')192        api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),193                                            str(aggregate.id),194                                            'host1')195        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \196                .AndReturn(aggregate)197        api.nova.host_list(IsA(http.HttpRequest)) \198                .AndReturn(self.hosts.list())199        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \200                .AndReturn(aggregate)201        api.nova.add_host_to_aggregate(IsA(http.HttpRequest),202                                       str(aggregate.id), host.host_name)203        self.mox.ReplayAll()204        res = self.client.post(reverse(constants.AGGREGATES_MANAGE_HOSTS_URL,205                                       args=[aggregate.id]),206                               form_data)207        self.assertNoFormErrors(res)208        self.assertRedirectsNoFollow(res,209                                     reverse(constants.AGGREGATES_INDEX_URL))210    @test.create_stubs({api.nova: ('aggregate_get', 'add_host_to_aggregate',211                                   'remove_host_from_aggregate',212                                   'host_list')})213    def test_manage_hosts_update_add_not_empty_aggregate_should_fail(self):214        aggregate = self.aggregates.first()215        aggregate.hosts = ['devstack001']216        host1 = self.hosts.list()[0]217        host3 = self.hosts.list()[2]218        form_data = {'manageaggregatehostsaction_role_member':219                     [host1.host_name, host3.host_name]}220        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \221                .AndReturn(aggregate)222        api.nova.host_list(IsA(http.HttpRequest)) \223                .AndReturn(self.hosts.list())224        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \225                .AndReturn(aggregate)226        api.nova.add_host_to_aggregate(IsA(http.HttpRequest),227                                       str(aggregate.id), host3.host_name)\228                .AndRaise(self.exceptions.nova)229        self.mox.ReplayAll()230        res = self.client.post(reverse(constants.AGGREGATES_MANAGE_HOSTS_URL,231                                       args=[aggregate.id]),232                               form_data)233        self.assertNoFormErrors(res)234        self.assertMessageCount(error=2)235        self.assertRedirectsNoFollow(res,236                                     reverse(constants.AGGREGATES_INDEX_URL))237    @test.create_stubs({api.nova: ('aggregate_get', 'add_host_to_aggregate',238                                   'remove_host_from_aggregate',239                                   'host_list')})240    def test_manage_hosts_update_clean_not_empty_aggregate_should_fail(self):241        aggregate = self.aggregates.first()242        aggregate.hosts = ['host1', 'host2', 'host3']243        form_data = {'manageaggregatehostsaction_role_member':244                     []}245        api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),246                                            str(aggregate.id),247                                            'host3')248        api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),249                                            str(aggregate.id),250                                            'host2')\251                .AndRaise(self.exceptions.nova)252        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \253                .AndReturn(aggregate)254        api.nova.host_list(IsA(http.HttpRequest)) \255                .AndReturn(self.hosts.list())256        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \257                .AndReturn(aggregate)258        self.mox.ReplayAll()259        res = self.client.post(reverse(constants.AGGREGATES_MANAGE_HOSTS_URL,260                                       args=[aggregate.id]),261                               form_data)262        self.assertNoFormErrors(res)263        self.assertMessageCount(error=2)264        self.assertRedirectsNoFollow(res,265                                     reverse(constants.AGGREGATES_INDEX_URL))266    @test.create_stubs({api.nova: ('aggregate_get', 'add_host_to_aggregate',267                                   'remove_host_from_aggregate',268                                   'host_list')})269    def _test_manage_hosts_update(self,270                                  host,271                                  aggregate,272                                  form_data,273                                  addAggregate=False,274                                  cleanAggregates=False):275        if cleanAggregates:276            api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),277                                                str(aggregate.id),278                                                'host3')279            api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),280                                                str(aggregate.id),281                                                'host2')282            api.nova.remove_host_from_aggregate(IsA(http.HttpRequest),283                                                str(aggregate.id),284                                                'host1')285        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \286                .AndReturn(aggregate)287        api.nova.host_list(IsA(http.HttpRequest)) \288                .AndReturn(self.hosts.list())289        api.nova.aggregate_get(IsA(http.HttpRequest), str(aggregate.id)) \290                .AndReturn(aggregate)291        if addAggregate:292            api.nova.add_host_to_aggregate(IsA(http.HttpRequest),293                                           str(aggregate.id),294                                           host.host_name)295        self.mox.ReplayAll()296        res = self.client.post(reverse(constants.AGGREGATES_MANAGE_HOSTS_URL,297                                       args=[aggregate.id]),298                               form_data)299        self.assertNoFormErrors(res)300        self.assertRedirectsNoFollow(res,301                                     reverse(constants.AGGREGATES_INDEX_URL))302    def test_manage_hosts_update_nothing_not_empty_aggregate(self):303        aggregate = self.aggregates.first()304        host = self.hosts.list()[0]305        aggregate.hosts = [host.host_name]306        form_data = {'manageaggregatehostsaction_role_member':307                     [host.host_name]}308        self._test_manage_hosts_update(host,309                                         aggregate,310                                         form_data,311                                         addAggregate=False)312    def test_manage_hosts_update_nothing_empty_aggregate(self):313        aggregate = self.aggregates.first()314        aggregate.hosts = []315        form_data = {'manageaggregatehostsaction_role_member':316                     []}317        self._test_manage_hosts_update(None,318                                         aggregate,319                                         form_data,320                                         addAggregate=False)321    def test_manage_hosts_update_add_empty_aggregate(self):322        aggregate = self.aggregates.first()323        aggregate.hosts = []324        host = self.hosts.list()[0]325        form_data = {'manageaggregatehostsaction_role_member':326                     [host.host_name]}327        self._test_manage_hosts_update(host,328                                         aggregate,329                                         form_data,330                                         addAggregate=True)331    def test_manage_hosts_update_add_not_empty_aggregate(self):332        aggregate = self.aggregates.first()333        aggregate.hosts = ['devstack001']334        host1 = self.hosts.list()[0]335        host3 = self.hosts.list()[2]336        form_data = {'manageaggregatehostsaction_role_member':337                     [host1.host_name, host3.host_name]}338        self._test_manage_hosts_update(host3,339                                         aggregate,340                                         form_data,341                                         addAggregate=True)342    def test_manage_hosts_update_clean_not_empty_aggregate(self):343        aggregate = self.aggregates.first()344        aggregate.hosts = ['host1', 'host2', 'host3']345        form_data = {'manageaggregatehostsaction_role_member':346                     []}347        self._test_manage_hosts_update(None,348                                         aggregate,349                                         form_data,350                                         addAggregate=False,...

test_aggregates.py

Source:test_aggregates.py

...14        AggregateTestModel.objects.create(boolean_field=False, char_field='Foo2', integer_field=1)15        AggregateTestModel.objects.create(boolean_field=False, char_field='Foo3', integer_field=2)16        AggregateTestModel.objects.create(boolean_field=True, char_field='Foo4', integer_field=0)17    def test_array_agg_charfield(self):18        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('char_field'))19        self.assertEqual(values, {'arrayagg': ['Foo1', 'Foo2', 'Foo3', 'Foo4']})20    def test_array_agg_integerfield(self):21        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('integer_field'))22        self.assertEqual(values, {'arrayagg': [0, 1, 2, 0]})23    def test_array_agg_booleanfield(self):24        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('boolean_field'))25        self.assertEqual(values, {'arrayagg': [True, False, False, True]})26    def test_array_agg_empty_result(self):27        AggregateTestModel.objects.all().delete()28        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('char_field'))29        self.assertEqual(values, {'arrayagg': []})30        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('integer_field'))31        self.assertEqual(values, {'arrayagg': []})32        values = AggregateTestModel.objects.aggregate(arrayagg=ArrayAgg('boolean_field'))33        self.assertEqual(values, {'arrayagg': []})34    def test_bit_and_general(self):35        values = AggregateTestModel.objects.filter(36            integer_field__in=[0, 1]).aggregate(bitand=BitAnd('integer_field'))37        self.assertEqual(values, {'bitand': 0})38    def test_bit_and_on_only_true_values(self):39        values = AggregateTestModel.objects.filter(40            integer_field=1).aggregate(bitand=BitAnd('integer_field'))41        self.assertEqual(values, {'bitand': 1})42    def test_bit_and_on_only_false_values(self):43        values = AggregateTestModel.objects.filter(44            integer_field=0).aggregate(bitand=BitAnd('integer_field'))45        self.assertEqual(values, {'bitand': 0})46    def test_bit_and_empty_result(self):47        AggregateTestModel.objects.all().delete()48        values = AggregateTestModel.objects.aggregate(bitand=BitAnd('integer_field'))49        self.assertEqual(values, {'bitand': None})50    def test_bit_or_general(self):51        values = AggregateTestModel.objects.filter(52            integer_field__in=[0, 1]).aggregate(bitor=BitOr('integer_field'))53        self.assertEqual(values, {'bitor': 1})54    def test_bit_or_on_only_true_values(self):55        values = AggregateTestModel.objects.filter(56            integer_field=1).aggregate(bitor=BitOr('integer_field'))57        self.assertEqual(values, {'bitor': 1})58    def test_bit_or_on_only_false_values(self):59        values = AggregateTestModel.objects.filter(60            integer_field=0).aggregate(bitor=BitOr('integer_field'))61        self.assertEqual(values, {'bitor': 0})62    def test_bit_or_empty_result(self):63        AggregateTestModel.objects.all().delete()64        values = AggregateTestModel.objects.aggregate(bitor=BitOr('integer_field'))65        self.assertEqual(values, {'bitor': None})66    def test_bool_and_general(self):67        values = AggregateTestModel.objects.aggregate(booland=BoolAnd('boolean_field'))68        self.assertEqual(values, {'booland': False})69    def test_bool_and_empty_result(self):70        AggregateTestModel.objects.all().delete()71        values = AggregateTestModel.objects.aggregate(booland=BoolAnd('boolean_field'))72        self.assertEqual(values, {'booland': None})73    def test_bool_or_general(self):74        values = AggregateTestModel.objects.aggregate(boolor=BoolOr('boolean_field'))75        self.assertEqual(values, {'boolor': True})76    def test_bool_or_empty_result(self):77        AggregateTestModel.objects.all().delete()78        values = AggregateTestModel.objects.aggregate(boolor=BoolOr('boolean_field'))79        self.assertEqual(values, {'boolor': None})80    def test_string_agg_requires_delimiter(self):81        with self.assertRaises(TypeError):82            AggregateTestModel.objects.aggregate(stringagg=StringAgg('char_field'))83    def test_string_agg_charfield(self):84        values = AggregateTestModel.objects.aggregate(stringagg=StringAgg('char_field', delimiter=';'))85        self.assertEqual(values, {'stringagg': 'Foo1;Foo2;Foo3;Foo4'})86    def test_string_agg_empty_result(self):87        AggregateTestModel.objects.all().delete()88        values = AggregateTestModel.objects.aggregate(stringagg=StringAgg('char_field', delimiter=';'))89        self.assertEqual(values, {'stringagg': ''})90class TestStatisticsAggregate(PostgresSQLTestCase):91    @classmethod92    def setUpTestData(cls):93        StatTestModel.objects.create(94            int1=1,95            int2=3,96            related_field=AggregateTestModel.objects.create(integer_field=0),97        )98        StatTestModel.objects.create(99            int1=2,100            int2=2,101            related_field=AggregateTestModel.objects.create(integer_field=1),102        )103        StatTestModel.objects.create(104            int1=3,105            int2=1,106            related_field=AggregateTestModel.objects.create(integer_field=2),107        )108    # Tests for base class (StatAggregate)109    def test_missing_arguments_raises_exception(self):110        with self.assertRaisesMessage(ValueError, 'Both y and x must be provided.'):111            StatAggregate(x=None, y=None)112    def test_correct_source_expressions(self):113        func = StatAggregate(x='test', y=13)114        self.assertIsInstance(func.source_expressions[0], Value)115        self.assertIsInstance(func.source_expressions[1], F)116    def test_alias_is_required(self):117        class SomeFunc(StatAggregate):118            function = 'TEST'119        with self.assertRaisesMessage(TypeError, 'Complex aggregates require an alias'):120            StatTestModel.objects.aggregate(SomeFunc(y='int2', x='int1'))121    # Test aggregates122    def test_corr_general(self):123        values = StatTestModel.objects.aggregate(corr=Corr(y='int2', x='int1'))124        self.assertEqual(values, {'corr': -1.0})125    def test_corr_empty_result(self):126        StatTestModel.objects.all().delete()127        values = StatTestModel.objects.aggregate(corr=Corr(y='int2', x='int1'))128        self.assertEqual(values, {'corr': None})129    def test_covar_pop_general(self):130        values = StatTestModel.objects.aggregate(covarpop=CovarPop(y='int2', x='int1'))131        self.assertEqual(values, {'covarpop': Approximate(-0.66, places=1)})132    def test_covar_pop_empty_result(self):133        StatTestModel.objects.all().delete()134        values = StatTestModel.objects.aggregate(covarpop=CovarPop(y='int2', x='int1'))135        self.assertEqual(values, {'covarpop': None})136    def test_covar_pop_sample(self):137        values = StatTestModel.objects.aggregate(covarpop=CovarPop(y='int2', x='int1', sample=True))138        self.assertEqual(values, {'covarpop': -1.0})139    def test_covar_pop_sample_empty_result(self):140        StatTestModel.objects.all().delete()141        values = StatTestModel.objects.aggregate(covarpop=CovarPop(y='int2', x='int1', sample=True))142        self.assertEqual(values, {'covarpop': None})143    def test_regr_avgx_general(self):144        values = StatTestModel.objects.aggregate(regravgx=RegrAvgX(y='int2', x='int1'))145        self.assertEqual(values, {'regravgx': 2.0})146    def test_regr_avgx_empty_result(self):147        StatTestModel.objects.all().delete()148        values = StatTestModel.objects.aggregate(regravgx=RegrAvgX(y='int2', x='int1'))149        self.assertEqual(values, {'regravgx': None})150    def test_regr_avgy_general(self):151        values = StatTestModel.objects.aggregate(regravgy=RegrAvgY(y='int2', x='int1'))152        self.assertEqual(values, {'regravgy': 2.0})153    def test_regr_avgy_empty_result(self):154        StatTestModel.objects.all().delete()155        values = StatTestModel.objects.aggregate(regravgy=RegrAvgY(y='int2', x='int1'))156        self.assertEqual(values, {'regravgy': None})157    def test_regr_count_general(self):158        values = StatTestModel.objects.aggregate(regrcount=RegrCount(y='int2', x='int1'))159        self.assertEqual(values, {'regrcount': 3})160    def test_regr_count_empty_result(self):161        StatTestModel.objects.all().delete()162        values = StatTestModel.objects.aggregate(regrcount=RegrCount(y='int2', x='int1'))163        self.assertEqual(values, {'regrcount': 0})164    def test_regr_intercept_general(self):165        values = StatTestModel.objects.aggregate(regrintercept=RegrIntercept(y='int2', x='int1'))166        self.assertEqual(values, {'regrintercept': 4})167    def test_regr_intercept_empty_result(self):168        StatTestModel.objects.all().delete()169        values = StatTestModel.objects.aggregate(regrintercept=RegrIntercept(y='int2', x='int1'))170        self.assertEqual(values, {'regrintercept': None})171    def test_regr_r2_general(self):172        values = StatTestModel.objects.aggregate(regrr2=RegrR2(y='int2', x='int1'))173        self.assertEqual(values, {'regrr2': 1})174    def test_regr_r2_empty_result(self):175        StatTestModel.objects.all().delete()176        values = StatTestModel.objects.aggregate(regrr2=RegrR2(y='int2', x='int1'))177        self.assertEqual(values, {'regrr2': None})178    def test_regr_slope_general(self):179        values = StatTestModel.objects.aggregate(regrslope=RegrSlope(y='int2', x='int1'))180        self.assertEqual(values, {'regrslope': -1})181    def test_regr_slope_empty_result(self):182        StatTestModel.objects.all().delete()183        values = StatTestModel.objects.aggregate(regrslope=RegrSlope(y='int2', x='int1'))184        self.assertEqual(values, {'regrslope': None})185    def test_regr_sxx_general(self):186        values = StatTestModel.objects.aggregate(regrsxx=RegrSXX(y='int2', x='int1'))187        self.assertEqual(values, {'regrsxx': 2.0})188    def test_regr_sxx_empty_result(self):189        StatTestModel.objects.all().delete()190        values = StatTestModel.objects.aggregate(regrsxx=RegrSXX(y='int2', x='int1'))191        self.assertEqual(values, {'regrsxx': None})192    def test_regr_sxy_general(self):193        values = StatTestModel.objects.aggregate(regrsxy=RegrSXY(y='int2', x='int1'))194        self.assertEqual(values, {'regrsxy': -2.0})195    def test_regr_sxy_empty_result(self):196        StatTestModel.objects.all().delete()197        values = StatTestModel.objects.aggregate(regrsxy=RegrSXY(y='int2', x='int1'))198        self.assertEqual(values, {'regrsxy': None})199    def test_regr_syy_general(self):200        values = StatTestModel.objects.aggregate(regrsyy=RegrSYY(y='int2', x='int1'))201        self.assertEqual(values, {'regrsyy': 2.0})202    def test_regr_syy_empty_result(self):203        StatTestModel.objects.all().delete()204        values = StatTestModel.objects.aggregate(regrsyy=RegrSYY(y='int2', x='int1'))205        self.assertEqual(values, {'regrsyy': None})206    def test_regr_avgx_with_related_obj_and_number_as_argument(self):207        """208        This is more complex test to check if JOIN on field and209        number as argument works as expected.210        """211        values = StatTestModel.objects.aggregate(complex_regravgx=RegrAvgX(y=5, x='related_field__integer_field'))...

profiling_test.py

Source:profiling_test.py

1# Copyright 2017 The TensorFlow Authors. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14# ==============================================================================15"""Unit tests for the basic data structures and algorithms for profiling."""16from __future__ import absolute_import17from __future__ import division18from __future__ import print_function19from tensorflow.core.framework import step_stats_pb220from tensorflow.python.debug.lib import profiling21from tensorflow.python.framework import test_util22from tensorflow.python.platform import googletest23class AggregateProfile(test_util.TensorFlowTestCase):24  def setUp(self):25    node_1 = step_stats_pb2.NodeExecStats(26        node_name="Add/123",27        op_start_rel_micros=3,28        op_end_rel_micros=5,29        all_end_rel_micros=4)30    self.profile_datum_1 = profiling.ProfileDatum(31        "cpu:0", node_1, "/foo/bar.py", 10, "func1", "Add")32    node_2 = step_stats_pb2.NodeExecStats(33        node_name="Mul/456",34        op_start_rel_micros=13,35        op_end_rel_micros=16,36        all_end_rel_micros=17)37    self.profile_datum_2 = profiling.ProfileDatum(38        "cpu:0", node_2, "/foo/bar.py", 11, "func1", "Mul")39    node_3 = step_stats_pb2.NodeExecStats(40        node_name="Add/123",41        op_start_rel_micros=103,42        op_end_rel_micros=105,43        all_end_rel_micros=4)44    self.profile_datum_3 = profiling.ProfileDatum(45        "cpu:0", node_3, "/foo/bar.py", 12, "func1", "Add")46    node_4 = step_stats_pb2.NodeExecStats(47        node_name="Add/123",48        op_start_rel_micros=203,49        op_end_rel_micros=205,50        all_end_rel_micros=4)51    self.profile_datum_4 = profiling.ProfileDatum(52        "gpu:0", node_4, "/foo/bar.py", 13, "func1", "Add")53  def testAggregateProfileConstructorWorks(self):54    aggregate_data = profiling.AggregateProfile(self.profile_datum_1)55    self.assertEqual(2, aggregate_data.total_op_time)56    self.assertEqual(4, aggregate_data.total_exec_time)57    self.assertEqual(1, aggregate_data.node_count)58    self.assertEqual(1, aggregate_data.node_exec_count)59  def testAddToAggregateProfileWithDifferentNodeWorks(self):60    aggregate_data = profiling.AggregateProfile(self.profile_datum_1)61    aggregate_data.add(self.profile_datum_2)62    self.assertEqual(5, aggregate_data.total_op_time)63    self.assertEqual(21, aggregate_data.total_exec_time)64    self.assertEqual(2, aggregate_data.node_count)65    self.assertEqual(2, aggregate_data.node_exec_count)66  def testAddToAggregateProfileWithSameNodeWorks(self):67    aggregate_data = profiling.AggregateProfile(self.profile_datum_1)68    aggregate_data.add(self.profile_datum_2)69    aggregate_data.add(self.profile_datum_3)70    self.assertEqual(7, aggregate_data.total_op_time)71    self.assertEqual(25, aggregate_data.total_exec_time)72    self.assertEqual(2, aggregate_data.node_count)73    self.assertEqual(3, aggregate_data.node_exec_count)74  def testAddToAggregateProfileWithDifferentDeviceSameNodeWorks(self):75    aggregate_data = profiling.AggregateProfile(self.profile_datum_1)76    aggregate_data.add(self.profile_datum_4)77    self.assertEqual(4, aggregate_data.total_op_time)78    self.assertEqual(8, aggregate_data.total_exec_time)79    self.assertEqual(2, aggregate_data.node_count)80    self.assertEqual(2, aggregate_data.node_exec_count)81if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.