openstack
/
octavia
Code Issues Proposed changes

Support to filter resources by tags 

Add 'tags', 'tags-any', 'not-tags', 'not-tags-any' param keys for
resource query.

Doc reference:
http://specs.openstack.org/openstack/api-wg/guidelines/tags.html#filtering-and-searching-by-tags

Change-Id: Ie752aa8e744d54c4909b97876dee450293d29908
Story: 2004810
Task: 28970
This commit is contained in:
Lingxian Kong 2019-01-21 17:08:04 +13:00
parent d0c499dbd6
commit 5df9ad49b3
8 changed files with 709 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
octavia/api/common/pagination.py
View File

@ -17,12 +17,14 @@ import copy
from oslo_log import log as logging
from pecan import request
import sqlalchemy
from sqlalchemy.orm import aliased
import sqlalchemy.sql as sa_sql
from octavia.api.common import types
from octavia.common.config import cfg
from octavia.common import constants
from octavia.common import exceptions
from octavia.db import base_models
from octavia.db import models
CONF = cfg.CONF
@ -178,6 +180,54 @@ class PaginationHelper(object):
links = [types.PageType(**link) for link in links]
return links
def _apply_tags_filtering(self, params, model, query):
if not getattr(model, "_tags", None):
return query
tag_alias = aliased(base_models.Tags)
if 'tags' in params:
tags = params.pop('tags')
if not isinstance(tags, list):
tags = [tags]
first_tag = tags.pop(0)
query = query.join(model._tags)
query = query.filter(base_models.Tags.tag == first_tag)
for tag in tags:
query = query.join(tag_alias, model._tags)
query = query.filter(tag_alias.tag == tag)
if 'tags-any' in params:
tags = params.pop('tags-any')
if not isinstance(tags, list):
tags = [tags]
query = query.join(tag_alias, model._tags)
query = query.filter(tag_alias.tag.in_(tags))
if 'not-tags' in params:
tags = params.pop('not-tags')
if not isinstance(tags, list):
tags = [tags]
first_tag = tags.pop(0)
subq = query.session.query(base_models.Tags.resource_id)
subq = subq.join(model._tags)
subq = subq.filter(base_models.Tags.tag == first_tag)
for tag in tags:
subq = subq.join(tag_alias, model._tags)
subq = subq.filter(tag_alias.tag == tag)
query = query.filter(~model.id.in_(subq))
if 'not-tags-any' in params:
tags = params.pop('not-tags-any')
if not isinstance(tags, list):
tags = [tags]
query = query.filter(
~model._tags.any(base_models.Tags.tag.in_(tags)))
return query
def apply(self, query, model, enforce_valid_params=True):
"""Returns a query with sorting / pagination criteria added.
@ -220,6 +270,9 @@ class PaginationHelper(object):
filter_params['load_balancer_id'] = filter_params.pop(
'loadbalancer_id')
# Apply tags filtering for the models which support tags.
query = self._apply_tags_filtering(filter_params, model, query)
# Drop invalid arguments
self.filters = {k: v for (k, v) in filter_params.items()
if k in vars(model.__data_model__())}

102
octavia/tests/functional/api/v2/test_health_monitor.py
View File

@ -611,6 +611,108 @@ class TestHealthMonitor(base.BaseAPITest):
self.assertEqual(hm1['id'],
hms['healthmonitors'][0]['id'])
def test_get_all_tags_filter(self):
pool1 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool1').get('pool')
self.set_lb_status(self.lb_id)
pool2 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool2').get('pool')
self.set_lb_status(self.lb_id)
pool3 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool3').get('pool')
self.set_lb_status(self.lb_id)
hm1 = self.create_health_monitor(
pool1.get('id'), constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1, tags=['test_tag1', 'test_tag2']).get(self.root_tag)
self.set_lb_status(self.lb_id)
hm2 = self.create_health_monitor(
pool2.get('id'), constants.HEALTH_MONITOR_PING,
1, 1, 1, 1, tags=['test_tag2', 'test_tag3']).get(self.root_tag)
self.set_lb_status(self.lb_id)
hm3 = self.create_health_monitor(
pool3.get('id'), constants.HEALTH_MONITOR_TCP,
1, 1, 1, 1, tags=['test_tag4', 'test_tag5']).get(self.root_tag)
self.set_lb_status(self.lb_id)
hms = self.get(
self.HMS_PATH,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(2, len(hms))
self.assertEqual(
[hm1.get('id'), hm2.get('id')],
[hm.get('id') for hm in hms]
)
hms = self.get(
self.HMS_PATH,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(1, len(hms))
self.assertEqual(
[hm2.get('id')],
[hm.get('id') for hm in hms]
)
hms = self.get(
self.HMS_PATH,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(2, len(hms))
self.assertEqual(
[hm1.get('id'), hm2.get('id')],
[hm.get('id') for hm in hms]
)
hms = self.get(
self.HMS_PATH,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(1, len(hms))
self.assertEqual(
[hm3.get('id')],
[hm.get('id') for hm in hms]
)
hms = self.get(
self.HMS_PATH,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(0, len(hms))
hms = self.get(
self.HMS_PATH,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(2, len(hms))
self.assertEqual(
[hm1.get('id'), hm2.get('id')],
[hm.get('id') for hm in hms]
)
hms = self.get(
self.HMS_PATH,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(hms, list)
self.assertEqual(0, len(hms))
def test_empty_get_all(self):
response = self.get(self.HMS_PATH).json.get(self.root_tag_list)
self.assertIsInstance(response, list)

86
octavia/tests/functional/api/v2/test_l7policy.py
View File

@ -501,6 +501,92 @@ class TestL7Policy(base.BaseAPITest):
self.assertEqual(policy1['id'],
l7pos['l7policies'][0]['id'])
def test_get_all_tags_filter(self):
policy1 = self.create_l7policy(
self.listener_id, constants.L7POLICY_ACTION_REJECT,
tags=['test_tag1', 'test_tag2']).get(self.root_tag)
self.set_lb_status(self.lb_id)
policy2 = self.create_l7policy(
self.listener_id, constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
position=2, redirect_pool_id=self.pool_id,
tags=['test_tag2', 'test_tag3']).get(self.root_tag)
self.set_lb_status(self.lb_id)
policy3 = self.create_l7policy(
self.listener_id, constants.L7POLICY_ACTION_REDIRECT_TO_URL,
redirect_url='http://localhost/',
tags=['test_tag4', 'test_tag5']).get(self.root_tag)
self.set_lb_status(self.lb_id)
policies = self.get(
self.L7POLICIES_PATH,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(2, len(policies))
self.assertEqual(
[policy1.get('id'), policy2.get('id')],
[policy.get('id') for policy in policies]
)
policies = self.get(
self.L7POLICIES_PATH,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(1, len(policies))
self.assertEqual(
[policy2.get('id')],
[policy.get('id') for policy in policies]
)
policies = self.get(
self.L7POLICIES_PATH,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(2, len(policies))
self.assertEqual(
[policy1.get('id'), policy2.get('id')],
[policy.get('id') for policy in policies]
)
policies = self.get(
self.L7POLICIES_PATH,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(1, len(policies))
self.assertEqual(
[policy3.get('id')],
[policy.get('id') for policy in policies]
)
policies = self.get(
self.L7POLICIES_PATH,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(0, len(policies))
policies = self.get(
self.L7POLICIES_PATH,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(2, len(policies))
self.assertEqual(
[policy1.get('id'), policy2.get('id')],
[policy.get('id') for policy in policies]
)
policies = self.get(
self.L7POLICIES_PATH,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(policies, list)
self.assertEqual(0, len(policies))
def test_empty_get_all(self):
response = self.get(self.L7POLICIES_PATH).json.get(self.root_tag_list)
self.assertIsInstance(response, list)

89
octavia/tests/functional/api/v2/test_l7rule.py
View File

@ -353,6 +353,95 @@ class TestL7Rule(base.BaseAPITest):
self.assertEqual(ru1['id'],
l7rus['rules'][0]['id'])
def test_get_all_tags_filter(self):
rule1 = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
constants.L7RULE_COMPARE_TYPE_STARTS_WITH,
'/api', tags=['test_tag1', 'test_tag2']).get(self.root_tag)
self.set_lb_status(self.lb_id)
rule2 = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_COOKIE,
constants.L7RULE_COMPARE_TYPE_CONTAINS, 'some-value',
key='some-cookie',
tags=['test_tag2', 'test_tag3']).get(self.root_tag)
self.set_lb_status(self.lb_id)
rule3 = self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_HOST_NAME,
constants.L7RULE_COMPARE_TYPE_EQUAL_TO,
'www.example.com',
tags=['test_tag4', 'test_tag5']).get(self.root_tag)
self.set_lb_status(self.lb_id)
rules = self.get(
self.l7rules_path,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(2, len(rules))
self.assertEqual(
[rule1.get('id'), rule2.get('id')],
[rule.get('id') for rule in rules]
)
rules = self.get(
self.l7rules_path,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(1, len(rules))
self.assertEqual(
[rule2.get('id')],
[rule.get('id') for rule in rules]
)
rules = self.get(
self.l7rules_path,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(2, len(rules))
self.assertEqual(
[rule1.get('id'), rule2.get('id')],
[rule.get('id') for rule in rules]
)
rules = self.get(
self.l7rules_path,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(1, len(rules))
self.assertEqual(
[rule3.get('id')],
[rule.get('id') for rule in rules]
)
rules = self.get(
self.l7rules_path,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(0, len(rules))
rules = self.get(
self.l7rules_path,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(2, len(rules))
self.assertEqual(
[rule1.get('id'), rule2.get('id')],
[rule.get('id') for rule in rules]
)
rules = self.get(
self.l7rules_path,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(rules, list)
self.assertEqual(0, len(rules))
def test_empty_get_all(self):
response = self.get(self.l7rules_path).json.get(self.root_tag_list)
self.assertIsInstance(response, list)

96
octavia/tests/functional/api/v2/test_listener.py
View File

@ -372,6 +372,102 @@ class TestListener(base.BaseAPITest):
self.assertEqual(li1['id'],
lis['listeners'][0]['id'])
def test_get_all_tags_filter(self):
listener1 = self.create_listener(
constants.PROTOCOL_HTTP,
80,
self.lb_id,
name='listener1',
tags=['test_tag1', 'test_tag2']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
listener2 = self.create_listener(
constants.PROTOCOL_HTTP,
81,
self.lb_id,
name='listener2',
tags=['test_tag2', 'test_tag3']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
listener3 = self.create_listener(
constants.PROTOCOL_HTTP,
82,
self.lb_id,
name='listener3',
tags=['test_tag4', 'test_tag5']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
listeners = self.get(
self.LISTENERS_PATH,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(2, len(listeners))
self.assertEqual(
[listener1.get('id'), listener2.get('id')],
[listener.get('id') for listener in listeners]
)
listeners = self.get(
self.LISTENERS_PATH,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(1, len(listeners))
self.assertEqual(
[listener2.get('id')],
[listener.get('id') for listener in listeners]
)
listeners = self.get(
self.LISTENERS_PATH,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(2, len(listeners))
self.assertEqual(
[listener1.get('id'), listener2.get('id')],
[listener.get('id') for listener in listeners]
)
listeners = self.get(
self.LISTENERS_PATH,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(1, len(listeners))
self.assertEqual(
[listener3.get('id')],
[listener.get('id') for listener in listeners]
)
listeners = self.get(
self.LISTENERS_PATH,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(0, len(listeners))
listeners = self.get(
self.LISTENERS_PATH,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(2, len(listeners))
self.assertEqual(
[listener1.get('id'), listener2.get('id')],
[listener.get('id') for listener in listeners]
)
listeners = self.get(
self.LISTENERS_PATH,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(listeners, list)
self.assertEqual(0, len(listeners))
def test_get_all_hides_deleted(self):
api_listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)

91
octavia/tests/functional/api/v2/test_load_balancer.py
View File

@ -1134,6 +1134,97 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertEqual(lb1['id'],
lbs['loadbalancers'][0]['id'])
def test_get_all_tags_filter(self):
lb1 = self.create_load_balancer(
uuidutils.generate_uuid(),
name='lb1',
project_id=self.project_id,
vip_address='10.0.0.1',
tags=['test_tag1', 'test_tag2']
).get(self.root_tag)
lb2 = self.create_load_balancer(
uuidutils.generate_uuid(),
name='lb2',
project_id=self.project_id,
tags=['test_tag2', 'test_tag3']
).get(self.root_tag)
lb3 = self.create_load_balancer(
uuidutils.generate_uuid(),
name='lb3',
project_id=self.project_id,
tags=['test_tag4', 'test_tag5']
).get(self.root_tag)
lbs = self.get(
self.LBS_PATH,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(2, len(lbs))
self.assertEqual(
[lb1.get('id'), lb2.get('id')],
[lb.get('id') for lb in lbs]
)
lbs = self.get(
self.LBS_PATH,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(1, len(lbs))
self.assertEqual(
[lb2.get('id')],
[lb.get('id') for lb in lbs]
)
lbs = self.get(
self.LBS_PATH,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(2, len(lbs))
self.assertEqual(
[lb1.get('id'), lb2.get('id')],
[lb.get('id') for lb in lbs]
)
lbs = self.get(
self.LBS_PATH,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(1, len(lbs))
self.assertEqual(
[lb3.get('id')],
[lb.get('id') for lb in lbs]
)
lbs = self.get(
self.LBS_PATH,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(0, len(lbs))
lbs = self.get(
self.LBS_PATH,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(2, len(lbs))
self.assertEqual(
[lb1.get('id'), lb2.get('id')],
[lb.get('id') for lb in lbs]
)
lbs = self.get(
self.LBS_PATH,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(lbs, list)
self.assertEqual(0, len(lbs))
def test_get_all_hides_deleted(self):
api_lb = self.create_load_balancer(
uuidutils.generate_uuid()).get(self.root_tag)

96
octavia/tests/functional/api/v2/test_member.py
View File

@ -345,6 +345,102 @@ class TestMember(base.BaseAPITest):
self.assertEqual(mem1['id'],
members['members'][0]['id'])
def test_get_all_tags_filter(self):
mem1 = self.create_member(
self.pool_id,
'192.0.2.1',
80,
name='member1',
tags=['test_tag1', 'test_tag2']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
mem2 = self.create_member(
self.pool_id,
'192.0.2.2',
80,
name='member2',
tags=['test_tag2', 'test_tag3']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
mem3 = self.create_member(
self.pool_id,
'192.0.2.3',
80,
name='member3',
tags=['test_tag4', 'test_tag5']
).get(self.root_tag)
self.set_lb_status(self.lb_id)
mems = self.get(
self.members_path,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(2, len(mems))
self.assertEqual(
[mem1.get('id'), mem2.get('id')],
[mem.get('id') for mem in mems]
)
mems = self.get(
self.members_path,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(1, len(mems))
self.assertEqual(
[mem2.get('id')],
[mem.get('id') for mem in mems]
)
mems = self.get(
self.members_path,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(2, len(mems))
self.assertEqual(
[mem1.get('id'), mem2.get('id')],
[mem.get('id') for mem in mems]
)
mems = self.get(
self.members_path,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(1, len(mems))
self.assertEqual(
[mem3.get('id')],
[mem.get('id') for mem in mems]
)
mems = self.get(
self.members_path,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(0, len(mems))
mems = self.get(
self.members_path,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(2, len(mems))
self.assertEqual(
[mem1.get('id'), mem2.get('id')],
[mem.get('id') for mem in mems]
)
mems = self.get(
self.members_path,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(mems, list)
self.assertEqual(0, len(mems))
def test_empty_get_all(self):
response = self.get(self.members_path).json.get(self.root_tag_list)
self.assertIsInstance(response, list)

96
octavia/tests/functional/api/v2/test_pool.py
View File

@ -547,6 +547,102 @@ class TestPool(base.BaseAPITest):
self.assertEqual(po1['id'],
pools['pools'][0]['id'])
def test_get_all_tags_filter(self):
po1 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool1',
tags=['test_tag1', 'test_tag2']
).get(self.root_tag)
self.set_lb_status(lb_id=self.lb_id)
po2 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool2',
tags=['test_tag2', 'test_tag3']
).get(self.root_tag)
self.set_lb_status(lb_id=self.lb_id)
po3 = self.create_pool(
self.lb_id,
constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN,
name='pool3',
tags=['test_tag4', 'test_tag5']
).get(self.root_tag)
self.set_lb_status(lb_id=self.lb_id)
pos = self.get(
self.POOLS_PATH,
params={'tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(2, len(pos))
self.assertEqual(
[po1.get('id'), po2.get('id')],
[po.get('id') for po in pos]
)
pos = self.get(
self.POOLS_PATH,
params={'tags': ['test_tag2', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(1, len(pos))
self.assertEqual(
[po2.get('id')],
[po.get('id') for po in pos]
)
pos = self.get(
self.POOLS_PATH,
params={'tags-any': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(2, len(pos))
self.assertEqual(
[po1.get('id'), po2.get('id')],
[po.get('id') for po in pos]
)
pos = self.get(
self.POOLS_PATH,
params={'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(1, len(pos))
self.assertEqual(
[po3.get('id')],
[po.get('id') for po in pos]
)
pos = self.get(
self.POOLS_PATH,
params={'not-tags-any': ['test_tag2', 'test_tag4']}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(0, len(pos))
pos = self.get(
self.POOLS_PATH,
params={'tags': 'test_tag2',
'tags-any': ['test_tag1', 'test_tag3']}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(2, len(pos))
self.assertEqual(
[po1.get('id'), po2.get('id')],
[po.get('id') for po in pos]
)
pos = self.get(
self.POOLS_PATH,
params={'tags': 'test_tag2', 'not-tags': 'test_tag2'}
).json.get(self.root_tag_list)
self.assertIsInstance(pos, list)
self.assertEqual(0, len(pos))
def test_empty_get_all(self):
response = self.get(self.POOLS_PATH).json.get(self.root_tag_list)
self.assertIsInstance(response, list)