docker/quay
1
0
Fork 0
mirror of https://github.com/quay/quay.git synced 2025-07-28 20:22:05 +03:00
Code Activity

Add additional detection for transitive deletes and fix those found (#281)

Browse Source 
We were not testing for transitive deletes when performing *User*
deletion, which led us to have a few
This commit is contained in:
Joseph Schorr
2020-03-25 15:58:12 -04:00
committed by GitHub
parent 3d2aadf91c
commit 59bc23313c
5 changed files with 61 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
test/test_api_usage.py
View File

@ -37,7 +37,7 @@ from data.appr_model.models import NEW_MODELS
from data.database import RepositoryActionCount, Repository as RepositoryTable
from data.logs_model import logs_model
from data.registry_model import registry_model
from test.helpers import assert_action_logged
from test.helpers import assert_action_logged, log_queries, check_transitive_modifications
from util.secscan.fake import fake_security_scanner
from endpoints.api.team import (
@ -2389,45 +2389,6 @@ class TestChangeRepoVisibility(ApiTestCase):
self.assertEquals(False, json["is_public"])
class log_queries(object):
def __init__(self, query_filters=None):
self.filters = query_filters
def get_queries(self):
queries = [q.msg[0] for q in self._handler.queries]
if not self.filters:
return queries
filtered_queries = []
for query_filter in self.filters:
filtered_queries.extend([q for q in queries if re.match(query_filter, q)])
return filtered_queries
def __enter__(self):
logger = logging.getLogger("peewee")
self._handler = _QueryLogHandler()
logger.setLevel(logging.DEBUG)
logger.addHandler(self._handler)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
logger = logging.getLogger("peewee")
logger.removeHandler(self._handler)
class check_transitive_modifications(log_queries):
def __init__(self):
filters = [r"^DELETE.+IN \(SELECT.+$", r"^UPDATE.+IN \(SELECT.+$"]
super(check_transitive_modifications, self).__init__(query_filters=filters)
def __exit__(self, exc_type, exc_val, exc_tb):
super(check_transitive_modifications, self).__exit__(exc_type, exc_val, exc_tb)
queries = self.get_queries()
if queries:
raise Exception("Detected transitive deletion or update in queries: %s" % queries)
class TestDeleteRepository(ApiTestCase):
SIMPLE_REPO = ADMIN_ACCESS_USER + "/simple"
COMPLEX_REPO = ADMIN_ACCESS_USER + "/complex"