<reponame>takanattie/osc-placement
# 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.

import uuid

from osc_placement.tests.functional import base


class TestAggregate(base.BaseTestCase):
    VERSION = '1.1'

    def test_fail_if_no_rp(self):
        self.assertCommandFailed(
            base.ARGUMENTS_MISSING,
            self.openstack,
            'resource provider aggregate list')

    def test_fail_if_rp_not_found(self):
        self.assertCommandFailed(
            'No resource provider',
            self.resource_provider_aggregate_list,
            'fake-uuid')

    def test_return_empty_list_if_no_aggregates(self):
        rp = self.resource_provider_create()
        self.assertEqual(
            [], self.resource_provider_aggregate_list(rp['uuid']))

    def test_success_set_aggregate(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs)

        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        self.resource_provider_aggregate_set(rp['uuid'])
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual([], rows)

    def test_set_aggregate_fail_if_no_rp(self):
        self.assertCommandFailed(
            base.ARGUMENTS_MISSING,
            self.openstack,
            'resource provider aggregate set')

    def test_success_set_multiple_aggregates(self):
        # each rp is associated with two aggregates
        rps = [self.resource_provider_create() for _ in range(2)]
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        for rp in rps:
            rows = self.resource_provider_aggregate_set(rp['uuid'], *aggs)
            self.assertEqual(aggs, {r['uuid'] for r in rows})
        # remove association for the first aggregate
        rows = self.resource_provider_aggregate_set(rps[0]['uuid'])
        self.assertEqual([], rows)
        # second rp should be in aggregates
        rows = self.resource_provider_aggregate_list(rps[1]['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        # cleanup
        rows = self.resource_provider_aggregate_set(rps[1]['uuid'])
        self.assertEqual([], rows)

    def test_success_set_large_number_aggregates(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(100)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs)
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_set(rp['uuid'])
        self.assertEqual([], rows)

    def test_fail_if_incorrect_aggregate_uuid(self):
        rp = self.resource_provider_create()
        aggs = ['abc', 'efg']
        self.assertCommandFailed(
            "is not a 'uuid'",
            self.resource_provider_aggregate_set,
            rp['uuid'], *aggs)

    # In version 1.1 a generation is not allowed.
    def test_fail_generation_arg_version_handling(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "Operation or argument is not supported with version 1.1",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation=5)


class TestAggregate119(TestAggregate):
    VERSION = '1.19'

    def test_success_set_aggregate(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs, generation=rp['generation'])

        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        self.resource_provider_aggregate_set(
            rp['uuid'], *[], generation=rp['generation'] + 1)
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual([], rows)

    def test_success_set_multiple_aggregates(self):
        # each rp is associated with two aggregates
        rps = [self.resource_provider_create() for _ in range(2)]
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        for rp in rps:
            rows = self.resource_provider_aggregate_set(
                rp['uuid'], *aggs, generation=rp['generation'])
            self.assertEqual(aggs, {r['uuid'] for r in rows})
        # remove association for the first aggregate
        rows = self.resource_provider_aggregate_set(
            rps[0]['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)
        # second rp should be in aggregates
        rows = self.resource_provider_aggregate_list(rps[1]['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        # cleanup
        rows = self.resource_provider_aggregate_set(
            rps[1]['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)

    def test_success_set_large_number_aggregates(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(100)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs, generation=rp['generation'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)

    def test_fail_incorrect_generation(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "Please update the generation and try again.",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation=99999)

    def test_fail_generation_not_int(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "invalid int value",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation='barney')

    def test_fail_if_incorrect_aggregate_uuid(self):
        rp = self.resource_provider_create()
        aggs = ['abc', 'efg']
        self.assertCommandFailed(
            "is not a 'uuid'",
            self.resource_provider_aggregate_set,
            rp['uuid'], *aggs, generation=rp['generation'])

    # In version 1.19 a generation is required.
    def test_fail_generation_arg_version_handling(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "A generation must be specified.",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg)

# 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.

import uuid

from osc_placement.tests.functional import base


class TestAggregate(base.BaseTestCase):
    VERSION = '1.1'

    def test_fail_if_no_rp(self):
        self.assertCommandFailed(
            base.ARGUMENTS_MISSING,
            self.openstack,
            'resource provider aggregate list')

    def test_fail_if_rp_not_found(self):
        self.assertCommandFailed(
            'No resource provider',
            self.resource_provider_aggregate_list,
            'fake-uuid')

    def test_return_empty_list_if_no_aggregates(self):
        rp = self.resource_provider_create()
        self.assertEqual(
            [], self.resource_provider_aggregate_list(rp['uuid']))

    def test_success_set_aggregate(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs)

        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        self.resource_provider_aggregate_set(rp['uuid'])
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual([], rows)

    def test_set_aggregate_fail_if_no_rp(self):
        self.assertCommandFailed(
            base.ARGUMENTS_MISSING,
            self.openstack,
            'resource provider aggregate set')

    def test_success_set_multiple_aggregates(self):
        # each rp is associated with two aggregates
        rps = [self.resource_provider_create() for _ in range(2)]
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        for rp in rps:
            rows = self.resource_provider_aggregate_set(rp['uuid'], *aggs)
            self.assertEqual(aggs, {r['uuid'] for r in rows})
        # remove association for the first aggregate
        rows = self.resource_provider_aggregate_set(rps[0]['uuid'])
        self.assertEqual([], rows)
        # second rp should be in aggregates
        rows = self.resource_provider_aggregate_list(rps[1]['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        # cleanup
        rows = self.resource_provider_aggregate_set(rps[1]['uuid'])
        self.assertEqual([], rows)

    def test_success_set_large_number_aggregates(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(100)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs)
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_set(rp['uuid'])
        self.assertEqual([], rows)

    def test_fail_if_incorrect_aggregate_uuid(self):
        rp = self.resource_provider_create()
        aggs = ['abc', 'efg']
        self.assertCommandFailed(
            "is not a 'uuid'",
            self.resource_provider_aggregate_set,
            rp['uuid'], *aggs)

    # In version 1.1 a generation is not allowed.
    def test_fail_generation_arg_version_handling(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "Operation or argument is not supported with version 1.1",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation=5)


class TestAggregate119(TestAggregate):
    VERSION = '1.19'

    def test_success_set_aggregate(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs, generation=rp['generation'])

        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        self.resource_provider_aggregate_set(
            rp['uuid'], *[], generation=rp['generation'] + 1)
        rows = self.resource_provider_aggregate_list(rp['uuid'])
        self.assertEqual([], rows)

    def test_success_set_multiple_aggregates(self):
        # each rp is associated with two aggregates
        rps = [self.resource_provider_create() for _ in range(2)]
        aggs = {str(uuid.uuid4()) for _ in range(2)}
        for rp in rps:
            rows = self.resource_provider_aggregate_set(
                rp['uuid'], *aggs, generation=rp['generation'])
            self.assertEqual(aggs, {r['uuid'] for r in rows})
        # remove association for the first aggregate
        rows = self.resource_provider_aggregate_set(
            rps[0]['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)
        # second rp should be in aggregates
        rows = self.resource_provider_aggregate_list(rps[1]['uuid'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        # cleanup
        rows = self.resource_provider_aggregate_set(
            rps[1]['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)

    def test_success_set_large_number_aggregates(self):
        rp = self.resource_provider_create()
        aggs = {str(uuid.uuid4()) for _ in range(100)}
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *aggs, generation=rp['generation'])
        self.assertEqual(aggs, {r['uuid'] for r in rows})
        rows = self.resource_provider_aggregate_set(
            rp['uuid'], *[], generation=rp['generation'] + 1)
        self.assertEqual([], rows)

    def test_fail_incorrect_generation(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "Please update the generation and try again.",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation=99999)

    def test_fail_generation_not_int(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "invalid int value",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg, generation='barney')

    def test_fail_if_incorrect_aggregate_uuid(self):
        rp = self.resource_provider_create()
        aggs = ['abc', 'efg']
        self.assertCommandFailed(
            "is not a 'uuid'",
            self.resource_provider_aggregate_set,
            rp['uuid'], *aggs, generation=rp['generation'])

    # In version 1.19 a generation is required.
    def test_fail_generation_arg_version_handling(self):
        rp = self.resource_provider_create()
        agg = str(uuid.uuid4())
        self.assertCommandFailed(
            "A generation must be specified.",
            self.resource_provider_aggregate_set,
            rp['uuid'], agg)
# 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. # each rp is associated with two aggregates # remove association for the first aggregate # second rp should be in aggregates # cleanup # In version 1.1 a generation is not allowed. # each rp is associated with two aggregates # remove association for the first aggregate # second rp should be in aggregates # cleanup # In version 1.19 a generation is required.