Merge branch 'master' into fix-enum-spelling

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 19.0.0
+current_version = 19.0.1
 commit = True
 tag = True
 tag_name = {new_version}

setup.py

@@ -52,7 +52,7 @@ def parse_reqs(req_path="./requirements.txt"):
 
 setup(
     name=NAME,
-    version="19.0.0",
+    version="19.0.1",
     description=DESCRIPTION,
     long_description=LONG_DESCRIPTION,
     long_description_content_type="text/markdown",

tests/conftest.py

@@ -20,13 +20,20 @@
     TrailblazerStatus,
 )
 from trailblazer.io.controller import ReadFile
+from trailblazer.store.models import Analysis
 
 
 @pytest.fixture(scope="session")
 def analysis_id_does_not_exist() -> int:
     return 123456789666
 
 
+@pytest.fixture(scope="session")
+def analysis_comment() -> str:
+    """Return a comment."""
+    return "a comment"
+
+
 @pytest.fixture(scope="session")
 def tower_id() -> str:
     """Return a NF Tower id."""
@@ -175,7 +182,7 @@ def analysis_store(
         store.add_user(name=user_data["name"], email=user_data["email"])
     for raw_analysis in raw_analyses:
         raw_analysis["user"] = store.get_user(email=raw_analysis["user"])
-        store.add(store.Analysis(**raw_analysis))
+        store.add(Analysis(**raw_analysis))
     store.commit()
     yield store
 

tests/store/filters/test_analysis_filters.py

@@ -8,6 +8,8 @@
     filter_analyses_by_case_id,
     filter_analyses_by_comment,
     filter_analyses_by_entry_id,
+    filter_analyses_by_is_visible,
+    filter_analyses_by_search_term,
     filter_analyses_by_started_at,
     filter_analyses_by_status,
     filter_analyses_by_statuses,
@@ -49,13 +51,13 @@ def test_filter_analyses_by_case_id(analysis_store: MockStore):
     assert existing_analysis == analysis.first()
 
 
-def test_filter_analyses_by_comment(analysis_store: MockStore):
+def test_filter_analyses_by_comment(analysis_store: MockStore, analysis_comment: str):
     """Test return analysis by comment when containing string."""
     # GIVEN a store containing analyses
     existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
 
     # GIVEN a comment
-    existing_analysis.comment = "a comment"
+    existing_analysis.comment = analysis_comment
 
     # WHEN retrieving an analysis by comment
     analysis: Query = filter_analyses_by_comment(
@@ -84,6 +86,138 @@ def test_filter_analyses_by_comment_when_not_matching(analysis_store: MockStore)
     assert not analysis.first()
 
 
+def test_filter_analyses_by_is_visible(analysis_store: MockStore):
+    """Test return analysis when is visible is true."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # GIVEN a visible analysis
+    existing_analysis.is_visible = True
+
+    # WHEN retrieving analyses by is visible
+    analyses: Query = filter_analyses_by_is_visible(
+        analyses=analysis_store.get_query(table=Analysis)
+    )
+
+    # THEN the analyses is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the analysis should match the original
+    assert existing_analysis == analyses.first()
+
+
+def test_filter_analyses_by_is_visible_when_false(analysis_store: MockStore):
+    """Test return analysis when is visible is false."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # GIVEN a hidden analysis
+    existing_analysis.is_visible = False
+
+    # WHEN retrieving analyses by is visible
+    analyses: Query = filter_analyses_by_is_visible(
+        analyses=analysis_store.get_query(table=Analysis)
+    )
+
+    # THEN the analyses is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the existing analysis should not be returned
+    for analysis in analyses:
+        assert existing_analysis != analysis
+
+
+def test_filter_analyses_by_search_term_in_comment(
+    analysis_store: MockStore, analysis_comment: str
+):
+    """Test return analysis when search term matches comment."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # GIVEN a a comment
+    existing_analysis.comment = analysis_comment
+
+    # WHEN retrieving analyses by search term
+    analyses: Query = filter_analyses_by_search_term(
+        analyses=analysis_store.get_query(table=Analysis), search_term=analysis_comment
+    )
+
+    # THEN the analysis is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the analysis should match the original
+    assert existing_analysis == analyses.first()
+
+
+def test_filter_analyses_by_search_term_in_data_analysis(analysis_store: MockStore):
+    """Test return analysis when search term matches data analysis."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # WHEN retrieving analyses by search term
+    analyses: Query = filter_analyses_by_search_term(
+        analyses=analysis_store.get_query(table=Analysis),
+        search_term=existing_analysis.data_analysis,
+    )
+
+    # THEN the analysis is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the analysis should match the original
+    assert existing_analysis == analyses.first()
+
+
+def test_filter_analyses_by_search_term_in_case(analysis_store: MockStore):
+    """Test return analysis when search term matches case."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # WHEN retrieving analyses by search term
+    analyses: Query = filter_analyses_by_search_term(
+        analyses=analysis_store.get_query(table=Analysis), search_term=existing_analysis.family
+    )
+
+    # THEN the analysis is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the analysis should match the original
+    assert existing_analysis == analyses.first()
+
+
+def test_filter_analyses_by_search_term_in_status(analysis_store: MockStore):
+    """Test return analysis when search term matches status."""
+    # GIVEN a store containing analyses
+    existing_analysis: Analysis = analysis_store.get_query(table=Analysis).first()
+
+    # WHEN retrieving analyses by search term
+    analyses: Query = filter_analyses_by_search_term(
+        analyses=analysis_store.get_query(table=Analysis), search_term=existing_analysis.status
+    )
+
+    # THEN the analyses is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the analysis should match the original
+    assert existing_analysis == analyses.first()
+
+
+def test_filter_analyses_by_search_term_when_not_matching(analysis_store: MockStore):
+    """Test return analysis when search term does not match."""
+    # GIVEN a store containing analyses
+    analysis_store.get_query(table=Analysis).first()
+
+    # WHEN retrieving analyses by search term
+    analyses: Query = filter_analyses_by_search_term(
+        analyses=analysis_store.get_query(table=Analysis),
+        search_term="does not exist",
+    )
+    # THEN the analyses is a query
+    assert isinstance(analyses, Query)
+
+    # THEN the no analysis is returned
+    assert not analyses.first()
+
+
 def test_filter_analyses_by_started_at(analysis_store: MockStore):
     """Test return analysis by started at when existing."""
     # GIVEN a store containing analyses

tests/store/test_base.py

@@ -1,7 +1,9 @@
 from datetime import datetime
+
 from sqlalchemy.orm import Query
 
 from tests.mocks.store_mock import MockStore
+from trailblazer.constants import TrailblazerStatus
 from trailblazer.store.models import Info, Job
 
 
@@ -35,3 +37,20 @@ def test_get_job_query_with_name_and_count_labels(job_store: MockStore):
     assert category.name == "0"
 
     assert category.count == job_store.get_query(table=Job).count()
+
+
+def test_get_analyses_query_by_search_term_and_is_visible(analysis_store: MockStore):
+    """Test return analyses query by search term and is visible"""
+
+    # GIVEN a store
+
+    # WHEN getting an object using the query
+    analyses: Query = analysis_store.get_analyses_query_by_search_term_and_is_visible(
+        search_term=TrailblazerStatus.PENDING
+    )
+
+    # THEN it should return a query
+    assert isinstance(analyses, Query)
+
+    # THEN it should return an object with the search term
+    assert analyses.first().status == TrailblazerStatus.PENDING

trailblazer/server/api.py

@@ -6,6 +6,7 @@
 
 from flask import Blueprint, Response, abort, g, jsonify, make_response, request
 from google.auth import jwt
+from sqlalchemy.orm import Query
 
 from trailblazer.constants import (
     ONE_MONTH_IN_DAYS,
@@ -54,18 +55,17 @@ def analyses():
     """Display analyses."""
     per_page = int(request.args.get("per_page", 100))
     page = int(request.args.get("page", 1))
-    query = store.analyses(
-        status=request.args.get("status"),
-        query=request.args.get("query"),
-        is_visible=request.args.get("is_visible") == "true" or None,
+    analyses: Query = store.get_analyses_query_by_search_term_and_is_visible(
+        search_term=request.args.get("query"),
+        is_visible=bool(request.args.get("is_visible")),
     )
 
-    query_page = query.paginate(page, per_page=per_page)
+    query_page = analyses.paginate(page, per_page=per_page)
     data = []
-    for analysis_obj in query_page.items:
-        analysis_data = analysis_obj.to_dict()
-        analysis_data["user"] = analysis_obj.user.to_dict() if analysis_obj.user else None
-        analysis_data["jobs"] = [job.to_dict() for job in analysis_obj.jobs]
+    for analysis in query_page.items:
+        analysis_data = analysis.to_dict()
+        analysis_data["user"] = analysis.user.to_dict() if analysis.user else None
+        analysis_data["jobs"] = [job.to_dict() for job in analysis.jobs]
         data.append(analysis_data)
     return jsonify(analyses=data)
 

trailblazer/store/api.py

@@ -1,76 +1,18 @@
 """Store backend in Trailblazer."""
-import datetime as dt
 import logging
 
 import alchy
-import sqlalchemy as sqa
-from alchy import Query
 
-from trailblazer.constants import TrailblazerStatus
 from trailblazer.store.core import CoreHandler
-from trailblazer.store.models import Analysis, Model
+from trailblazer.store.models import Model
 
 LOG = logging.getLogger(__name__)
 
 
 class BaseHandler(CoreHandler):
-    Analysis = Analysis
-
     def setup(self):
         self.create_all()
 
-    def analyses(
-        self,
-        case_id: str = None,
-        query: str = None,
-        status: str = None,
-        deleted: bool = None,
-        temp: bool = False,
-        before: dt.datetime = None,
-        is_visible: bool = None,
-        family: str = None,
-        data_analysis: str = None,
-        comment: str = None,
-    ) -> Query:
-        """
-        used by REST +> CG
-        Fetch analyses from the database."""
-        if not case_id:
-            case_id = family
-
-        analysis_query = self.Analysis.query
-        if case_id:
-            analysis_query = analysis_query.filter_by(family=case_id)
-        if query:  # to be deprecated
-            analysis_query = analysis_query.filter(
-                sqa.or_(
-                    self.Analysis.family.ilike(f"%{query}%"),
-                    self.Analysis.status.ilike(f"%{query}%"),
-                    self.Analysis.data_analysis.ilike(f"%{query}%"),
-                    self.Analysis.comment.ilike(f"%{query}%"),
-                )
-            )
-        if status:
-            analysis_query = analysis_query.filter_by(status=status)
-        if isinstance(deleted, bool):
-            analysis_query = analysis_query.filter_by(is_deleted=deleted)
-        if temp:
-            analysis_query = analysis_query.filter(
-                self.Analysis.status.in_(TrailblazerStatus.ongoing_statuses())
-            )
-        if before:
-            analysis_query = analysis_query.filter(self.Analysis.started_at < before)
-        if is_visible is not None:
-            analysis_query = analysis_query.filter_by(is_visible=is_visible)
-        if data_analysis:
-            analysis_query = analysis_query.filter(
-                Analysis.data_analysis.ilike(f"%{data_analysis}%")
-            )
-        if comment:
-            analysis_query = analysis_query.filter(Analysis.comment.ilike(f"%{comment}%"))
-
-        return analysis_query.order_by(self.Analysis.started_at.desc())
-
 
 class Store(alchy.Manager, BaseHandler):
     def __init__(self, uri: str):

trailblazer/store/base.py

@@ -1,9 +1,14 @@
 from dataclasses import dataclass
+from typing import Callable, Dict, List, Optional, Type, Union
+
 from sqlalchemy import func
 from sqlalchemy.orm import Query
-from typing import Type
 
-from trailblazer.store.models import Model, Job
+from trailblazer.store.filters.analyses_filters import (
+    AnalysisFilter,
+    apply_analysis_filter,
+)
+from trailblazer.store.models import Analysis, Job, Model
 
 
 @dataclass
@@ -23,3 +28,25 @@ def get_job_query_with_name_and_count_labels(self) -> Query:
             Job.name.label("name"),
             func.count(Job.id).label("count"),
         )
+
+    def get_analyses_query_by_search_term_and_is_visible(
+        self,
+        search_term: Optional[str] = None,
+        is_visible: bool = False,
+    ) -> Query:
+        """Return analyses by search term qnd is visible."""
+        if not search_term and not is_visible:
+            return
+        filter_map: Dict[Callable, Optional[Union[str, bool]]] = {
+            AnalysisFilter.FILTER_BY_IS_VISIBLE: is_visible,
+            AnalysisFilter.FILTER_BY_SEARCH_TERM: search_term,
+        }
+        filter_functions: List[Callable] = [
+            function for function, supplied_arg in filter_map.items() if supplied_arg
+        ]
+        analyses: Query = apply_analysis_filter(
+            filter_functions=filter_functions,
+            analyses=self.get_query(table=Analysis),
+            search_term=search_term,
+        )
+        return analyses.order_by(Analysis.started_at.desc())