def test_build_query_from_params(self):
        # Test with all default parameters
        config = self.get_dummy_context()
        params = {}
        params = scommon.get_parameters(params)
        query = ElasticSearchBase.build_query_from_params(params, config)
        self.assertTrue(query)
        self.assertTrue("query" in query)
        self.assertTrue("size" in query)
        self.assertTrue("from" in query)

        # Searching for a term in a specific field and with a specific product
        params = {
            "terms": "hang",
            "fields": "dump",
            "search_mode": "contains",
            "products": "fennec"
        }
        params = scommon.get_parameters(params)
        query = ElasticSearchBase.build_query_from_params(params, config)
        self.assertTrue(query)
        self.assertTrue("query" in query)
        self.assertTrue("filtered" in query["query"])

        filtered = query["query"]["filtered"]
        self.assertTrue("query" in filtered)
        self.assertTrue("wildcard" in filtered["query"])
        self.assertTrue("dump" in filtered["query"]["wildcard"])

        dump_term = filtered["query"]["wildcard"]["dump"]
        self.assertEqual(dump_term, "*hang*")
        self.assertTrue("filter" in filtered)
        self.assertTrue("and" in filtered["filter"])

    def test_get_parameters(self):
        """
        Test search_common.get_parameters()
        """
        # Empty params, only default values are returned
        params = co.get_parameters({})
        self.assertTrue(params)

        for i in params:
            typei = type(params[i])
            if i in ("from_date", "to_date", "build_from", "build_to"):
                self.assertTrue(typei is datetime)
            else:
                self.assertTrue(not params[i] or typei is int or typei is str
                                or typei is list)

        # Empty params
        params = co.get_parameters({
            "terms": "",
            "fields": "",
            "products": "",
            "from_date": "",
            "to_date": "",
            "versions": "",
            "reasons": "",
            "release_channels": "",
            "os": "",
            "search_mode": "",
            "build_ids": "",
            "report_process": "",
            "report_type": "",
            "plugin_in": "",
            "plugin_search_mode": "",
            "plugin_terms": ""
        })
        assert params, "SearchCommon.get_parameters() returned something " \
                       "empty or null."
        for i in params:
            typei = type(params[i])
            if i in ("from_date", "to_date", "build_from", "build_to"):
                self.assertTrue(typei is datetime)
            else:
                self.assertTrue(not params[i] or typei is int or typei is str
                                or typei is list)

        # Test with encoded slashes in terms and signature
        params = co.get_parameters({
            "terms": ["some", "terms/sig"],
            "signature": "my/little/signature"
        })

        self.assertTrue("signature" in params)
        self.assertTrue("terms" in params)
        self.assertEqual(params["terms"], ["some", "terms/sig"])
        self.assertEqual(params["signature"], "my/little/signature")

def test_get_parameters():
    """
    Test SearchCommon.get_parameters()
    """
    # Empty params, only default values are returned
    params = co.get_parameters({})
    assert params, (
            "SearchCommon.get_parameters() returned something empty or null.")
    for i in params:
        typei = type(params[i])
        if i in ("from_date", "to_date", "build_from", "build_to"):
            assert typei is datetime, (
                        "The parameter %s is of a non expected type %s, "
                        "should be datetime" % (i, typei))
        else:
            assert (not params[i] or typei is int or typei is str or
                    typei is list), (
                        "The parameter %s is of a non expected type %s" %
                        (i, typei))

    # Empty params
    params = co.get_parameters({
        "terms": "",
        "fields": "",
        "products": "",
        "from_date": "",
        "to_date": "",
        "versions": "",
        "reasons": "",
        "os": "",
        "branches": "",
        "search_mode": "",
        "build_ids": "",
        "report_process": "",
        "report_type": "",
        "plugin_in": "",
        "plugin_search_mode": "",
        "plugin_terms": ""
    })
    assert params, (
            "SearchCommon.get_parameters() returned something empty or null.")
    for i in params:
        typei = type(params[i])
        if i in ("from_date", "to_date", "build_from", "build_to"):
            assert typei is datetime, (
                        "The parameter %s is of a non expected type %s, "
                        "should be datetime" % (i, typei))
        else:
            assert (not params[i] or typei is int or typei is str or
                    typei is list), (
                        "The parameter %s is of a non expected type %s" %
                        (i, typei))

    def test_get_parameters(self):
        """
        Test search_common.get_parameters()
        """
        # Empty params, only default values are returned
        params = get_parameters({})
        assert params

        for i in params:
            typei = type(params[i])
            if i in ("from_date", "to_date", "build_from", "build_to"):
                assert typei is datetime.datetime
            else:
                assert not params[i] or typei in (int, str, list)

        # Empty params
        params = get_parameters({
            "terms": "",
            "fields": "",
            "products": "",
            "from_date": "",
            "to_date": "",
            "versions": "",
            "reasons": "",
            "release_channels": "",
            "os": "",
            "search_mode": "",
            "build_ids": "",
            "report_process": "",
            "report_type": "",
            "plugin_in": "",
            "plugin_search_mode": "",
            "plugin_terms": ""
        })
        assert params, "SearchCommon.get_parameters() returned something empty or null."
        for i in params:
            typei = type(params[i])
            if i in ("from_date", "to_date", "build_from", "build_to"):
                assert typei is datetime.datetime
            else:
                assert not params[i] or typei in (int, str, list)

        # Test with encoded slashes in terms and signature
        params = get_parameters({
            "terms": ["some", "terms/sig"],
            "signature": "my/little/signature"
        })

        assert "signature" in params
        assert "terms" in params
        assert params["terms"] == ["some", "terms/sig"]
        assert params["signature"] == "my/little/signature"

def test_build_query_from_params():
    """
    Test ElasticSearchBase.build_query_from_params()
    """
    # Test with all default parameters
    args = {
        "config": get_dummy_context()
    }
    search = ElasticSearchBase(**args)
    params = {}
    params = scommon.get_parameters(params)
    query = ElasticSearchBase.build_query_from_params(params)
    assert query, "build_query_from_params returned a bad value: %s" % query
    assert "query" in query, (
                "query is malformed, 'query' key missing: %s" % query)
    assert "size" in query, (
                "query is malformed, 'size' key missing: %s" % query)
    assert "from" in query, (
                "query is malformed, 'from' key missing: %s" % query)

    # Searching for a term in a specific field and with a specific product
    params = {
        "terms": "hang",
        "fields": "dump",
        "search_mode": "contains",
        "products": "fennec"
    }
    params = scommon.get_parameters(params)
    query = ElasticSearchBase.build_query_from_params(params)
    assert query, "build_query_from_params returned a bad value: %s" % query
    assert "query" in query, (
                "query is malformed, 'query' key missing: %s" % query)
    assert "filtered" in query["query"], (
                "query is malformed, 'filtered' key missing: %s" % query)

    filtered = query["query"]["filtered"]
    assert "query" in filtered, (
                "query is malformed,  'query' key missing: %s" % query)
    assert "wildcard" in filtered["query"], (
                "query is malformed, 'wildcard' key missing: %s" % query)
    assert "dump" in filtered["query"]["wildcard"], (
                "query is malformed, 'dump' key missing: %s" % query)

    dump_term = filtered["query"]["wildcard"]["dump"]
    assert "*hang*" == dump_term, (
                "query is malformed, value for wildcard is wrong: %s" % query)
    assert "filter" in filtered, (
                "query is malformed, 'filter' key missing: %s" % query)
    assert "and" in filtered["filter"], (
                "query is malformed, 'and' key missing: %s" % query)

    def get(self, **kwargs):
        """
        Search for crashes and return them.

        See http://socorro.readthedocs.org/en/latest/middleware.html#search

        Optional arguments: see SearchCommon.get_parameters()

        """
        # change aliases from the web to the implementation's need
        if "for" in kwargs and "terms" not in kwargs:
            kwargs["terms"] = kwargs.get("for")
        if "from" in kwargs and "from_date" not in kwargs:
            kwargs["from_date"] = kwargs.get("from")
        if "to" in kwargs and "to_date" not in kwargs:
            kwargs["to_date"] = kwargs.get("to")
        if "in" in kwargs and "fields" not in kwargs:
            kwargs["fields"] = kwargs.get("in")

        params = search_common.get_parameters(kwargs)

        # Get information about the versions
        versions_service = Util(config=self.context)
        params["versions_info"] = versions_service.versions_info(**params)

        query = Search.build_query_from_params(params, self.config)

        # For signatures mode, we need to collect more data with facets
        if params["data_type"] == "signatures":
            # No need to get crashes, we only want signatures
            query["size"] = 0
            query["from"] = 0

            # Using a fixed number instead of the needed number.
            # This hack limits the number of distinct signatures to process,
            # and hugely improves performances with long queries.

            query["facets"] = Search.get_signatures_facet(
                self.config.searchMaxNumberOfDistinctSignatures
            )

        json_query = json.dumps(query)
        logger.debug("Query the crashes or signatures: %s", json_query)

        es_result = self.query(params["from_date"],
                               params["to_date"],
                               json_query)

        # Executing the query and returning the result
        if params["data_type"] == "signatures":
            return self.search_for_signatures(params, es_result, query)
        else:
            return es_result

    def prepare_search_params(self, **kwargs):
        """Return a dictionary of parameters for a search-like SQL query.

        Uses socorro.lib.search_common.get_parameters() for arguments
        filtering.
        """
        params = search_common.get_parameters(kwargs)

        if not params["signature"]:
            raise MissingOrBadArgumentError(
                "Mandatory parameter 'signature' is missing or empty"
            )

        params["terms"] = params["signature"]
        params["search_mode"] = "is_exactly"

        # Default mode falls back to starts_with for postgres
        if params["plugin_search_mode"] == "default":
            params["plugin_search_mode"] = "starts_with"

        # Searching for terms in plugins
        if params["report_process"] == "plugin" and params["plugin_terms"]:
            params["plugin_terms"] = " ".join(params["plugin_terms"])
            params["plugin_terms"] = Crashes.prepare_terms(
                params["plugin_terms"],
                params["plugin_search_mode"]
            )

        # Get information about the versions
        util_service = Util(config=self.context)
        params["versions_info"] = util_service.versions_info(**params)

        # Parsing the versions
        params["versions_string"] = params["versions"]
        (params["versions"], params["products"]) = Crashes.parse_versions(
            params["versions"],
            params["products"]
        )

        # Changing the OS ids to OS names
        if hasattr(self.context, 'webapi'):
            context = self.context.webapi
        else:
            # old middleware
            context = self.context
        for i, elem in enumerate(params["os"]):
            for platform in context.platforms:
                if platform["id"] == elem:
                    params["os"][i] = platform["name"]

        return params

def test_build_reports_sql_where():
    """ Test PostgreSQLBase.build_reports_sql_where()."""
    pgbase = get_instance()
    params = search_common.get_parameters({}) # Get default search params
    sql_params = {}

    # Test 1: default values for parameters
    sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND %(to_date)s"
    sql_params_exp = {
        "from_date": params.from_date,
        "to_date": params.to_date
    }

    (sql, sql_params) = pgbase.build_reports_sql_where(params, sql_params)
    sql = " ".join(sql.split()) # squeeze all \s, \r, \t...

    assert sql == sql_exp, "Expected sql to be %s, got %s instead" % (sql_exp,
                                                                      sql)
    assert sql_params == sql_params_exp, "Expected sql params to be %s, got " \
                                         "%s instead" % (sql_params_exp,
                                                         sql_params)

    def test_build_reports_sql_where(self):
        """ Test PostgreSQLBase.build_reports_sql_where()."""
        config = self.get_dummy_context()
        pgbase = self.get_instance()
        params = search_common.get_parameters({})  # Get default search params
        default_params = util.DotDict(params.copy())
        sql_params = {}

        # .....................................................................
        # Test 1: default values for parameters
        sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND " \
                  "%(to_date)s"
        sql_params_exp = {
            "from_date": params.from_date,
            "to_date": params.to_date
        }

        (sql, sql_params) = pgbase.build_reports_sql_where(params, sql_params,
                                                           config)
        sql = " ".join(sql.split())  # squeeze all \s, \r, \t...

        self.assertEqual(sql, sql_exp)
        self.assertEqual(sql_params, sql_params_exp)

        # .....................................................................
        # Test 2: terms and search_mode = is_exactly
        sql_params = {}
        params.terms = "signature"
        params.search_mode = "is_exactly"

        sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND " \
                  "%(to_date)s AND r.signature=%(term)s"
        sql_params_exp = {
            "from_date": params.from_date,
            "to_date": params.to_date,
            "term": params.terms
        }

        (sql, sql_params) = pgbase.build_reports_sql_where(params, sql_params,
                                                           config)
        sql = " ".join(sql.split())  # squeeze all \s, \r, \t...

        self.assertEqual(sql, sql_exp)
        self.assertEqual(sql_params, sql_params_exp)

        # .....................................................................
        # Test 3: terms and search_mode != is_exactly
        sql_params = {}
        params.terms = "signature%"
        params.search_mode = "starts_with"

        sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND " \
                  "%(to_date)s AND r.signature LIKE %(term)s"
        sql_params_exp = {
            "from_date": params.from_date,
            "to_date": params.to_date,
            "term": params.terms
        }

        (sql, sql_params) = pgbase.build_reports_sql_where(params, sql_params,
                                                           config)
        sql = " ".join(sql.split())  # squeeze all \s, \r, \t...

        self.assertEqual(sql, sql_exp)
        self.assertEqual(sql_params, sql_params_exp)

        # .....................................................................
        # Test 4: products
        sql_params = {}
        params.terms = default_params.terms
        params.search_mode = default_params.search_mode
        params.products = ["Firefox", "Fennec"]

        sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND " \
                  "%(to_date)s AND (r.product=%(product0)s OR " \
                  "r.product=%(product1)s)"
        sql_params_exp = {
            "from_date": params.from_date,
            "to_date": params.to_date,
            "product0": "Firefox",
            "product1": "Fennec"
        }

        (sql, sql_params) = pgbase.build_reports_sql_where(params, sql_params,
                                                           config)
        sql = " ".join(sql.split())  # squeeze all \s, \r, \t...

        self.assertEqual(sql, sql_exp)
        self.assertEqual(sql_params, sql_params_exp)

        # .....................................................................
        # Test 5: os
        sql_params = {}
        params.products = default_params.products
        params.os = ["Windows"]

        sql_exp = "WHERE r.date_processed BETWEEN %(from_date)s AND " \
                  "%(to_date)s AND (r.os_name=%(os0)s)"
        sql_params_exp = {
            "from_date": params.from_date,
#.........这里部分代码省略.........

    def get_list(self, **kwargs):
        """
        List all crashes with a given signature and return them.

        Both `from_date` and `to_date` (and their aliases `from` and `to`)
        are required and can not be greater than 30 days apart.

        Optional arguments: see SearchCommon.get_parameters()

        """
        # aliases
        if "from" in kwargs and "from_date" not in kwargs:
            kwargs["from_date"] = kwargs.get("from")
        if "to" in kwargs and "to_date" not in kwargs:
            kwargs["to_date"] = kwargs.get("to")

        if not kwargs.get('from_date'):
            raise MissingArgumentError('from_date')
        if not kwargs.get('to_date'):
            raise MissingArgumentError('to_date')

        from_date = datetimeutil.datetimeFromISOdateString(kwargs['from_date'])
        to_date = datetimeutil.datetimeFromISOdateString(kwargs['to_date'])
        span_days = (to_date - from_date).days
        if span_days > 30:
            raise BadArgumentError(
                'Span between from_date and to_date can not be more than 30'
            )

        # start with the default
        sort_order = {
            'key': 'date_processed',
            'direction': 'DESC'
        }
        if 'sort' in kwargs:
            sort_order['key'] = kwargs.pop('sort')
            _recognized_sort_orders = (
                'date_processed',
                'uptime',
                'user_comments',
                'uuid',
                'uuid_text',
                'product',
                'version',
                'build',
                'signature',
                'url',
                'os_name',
                'os_version',
                'cpu_name',
                'cpu_info',
                'address',
                'reason',
                'last_crash',
                'install_age',
                'hangid',
                'process_type',
                'release_channel',
                'install_time',
                'duplicate_of',
            )
            if sort_order['key'] not in _recognized_sort_orders:
                raise BadArgumentError(
                    '%s is not a recognized sort order key' % sort_order['key']
                )
            sort_order['direction'] = 'ASC'
            if 'reverse' in kwargs:
                if kwargs.pop('reverse'):
                    sort_order['direction'] = 'DESC'

        include_raw_crash = kwargs.get('include_raw_crash') or False
        params = search_common.get_parameters(kwargs)

        if not params["signature"]:
            raise MissingArgumentError('signature')

        params["terms"] = params["signature"]
        params["search_mode"] = "is_exactly"

        # Default mode falls back to starts_with for postgres
        if params["plugin_search_mode"] == "default":
            params["plugin_search_mode"] = "starts_with"

        # Limiting to a signature
        if params["terms"]:
            params["terms"] = self.prepare_terms(params["terms"],
                                                 params["search_mode"])

        # Searching for terms in plugins
        if params["report_process"] == "plugin" and params["plugin_terms"]:
            params["plugin_terms"] = " ".join(params["plugin_terms"])
            params["plugin_terms"] = self.prepare_terms(
                params["plugin_terms"],
                params["plugin_search_mode"]
            )

        # Get information about the versions
        util_service = Util(config=self.context)
        params["versions_info"] = util_service.versions_info(**params)

#.........这里部分代码省略.........

    def get_list(self, **kwargs):
        """
        List all crashes with a given signature and return them.

        Optional arguments: see SearchCommon.get_parameters()

        """
        # Creating the connection to the DB
        self.connection = self.database.connection()
        cur = self.connection.cursor()

        params = search_common.get_parameters(kwargs)

        if params["signature"] is None:
            return None

        params["terms"] = params["signature"]
        params["search_mode"] = "is_exactly"

        # Default mode falls back to starts_with for postgres
        if params["plugin_search_mode"] == "default":
            params["plugin_search_mode"] = "starts_with"

        # Limiting to a signature
        if params["terms"]:
            params["terms"] = self.prepare_terms(params["terms"],
                                                 params["search_mode"])

        # Searching for terms in plugins
        if params["report_process"] == "plugin" and params["plugin_terms"]:
            params["plugin_terms"] = " ".join(params["plugin_terms"])
            params["plugin_terms"] = self.prepare_terms(
                                                params["plugin_terms"],
                                                params["plugin_search_mode"])

        # Get information about the versions
        util_service = Util(config=self.context)
        params["versions_info"] = util_service.versions_info(**params)

        # Parsing the versions
        params["versions_string"] = params["versions"]
        (params["versions"], params["products"]) = self.parse_versions(
                                                            params["versions"],
                                                            params["products"])

        # Changing the OS ids to OS names
        for i, elem in enumerate(params["os"]):
            for platform in self.context.platforms:
                if platform["id"] == elem:
                    params["os"][i] = platform["name"]

        # Creating the parameters for the sql query
        sql_params = {
        }

        # Preparing the different parts of the sql query

        sql_select = """
            SELECT
                r.date_processed,
                r.uptime,
                r.user_comments,
                r.uuid,
                r.product,
                r.version,
                r.build,
                r.signature,
                r.url,
                r.os_name,
                r.os_version,
                r.cpu_name,
                r.cpu_info,
                r.address,
                r.reason,
                r.last_crash,
                r.install_age,
                r.hangid,
                r.process_type,
                (r.client_crash_date - (r.install_age * INTERVAL '1 second'))
                    AS install_time,
                rd.duplicate_of
        """

        sql_from = self.build_reports_sql_from(params)
        sql_from = """%s
            LEFT OUTER JOIN reports_duplicates rd ON r.uuid = rd.uuid
        """ % sql_from

        (sql_where, sql_params) = self.build_reports_sql_where(params,
                                                               sql_params,
                                                               self.context)

        sql_order = """
            ORDER BY r.date_processed DESC
        """

        (sql_limit, sql_params) = self.build_reports_sql_limit(params,
                                                               sql_params)

        # Assembling the query
#.........这里部分代码省略.........

    def search(self, **kwargs):
        """
        Search for crashes and return them.

        See http://socorro.readthedocs.org/en/latest/middleware.html#search

        Optional arguments: see SearchCommon.get_parameters()

        """
        # Creating the connection to the DB
        self.connection = self.database.connection()
        cur = self.connection.cursor()

        params = search_common.get_parameters(kwargs)

        # Default mode falls back to starts_with for postgres
        if params["search_mode"] == "default":
            params["search_mode"] = "starts_with"
        if params["plugin_search_mode"] == "default":
            params["plugin_search_mode"] = "starts_with"

        # For Postgres, we never search for a list of terms
        if params["terms"]:
            params["terms"] = " ".join(params["terms"])
            params["terms"] = Search.prepare_terms(params["terms"],
                                                   params["search_mode"])

        # Searching for terms in plugins
        if params["report_process"] == "plugin" and params["plugin_terms"]:
            params["plugin_terms"] = " ".join(params["plugin_terms"])
            params["plugin_terms"] = Search.prepare_terms(
                                                params["plugin_terms"],
                                                params["plugin_search_mode"])

        # Parsing the versions
        params["versions_string"] = params["versions"]
        (params["versions"], params["products"]) = Search.parse_versions(
                                                            params["versions"],
                                                            params["products"])

        # Changing the OS ids to OS names
        for i, elem in enumerate(params["os"]):
            for platform in self.context.platforms:
                if platform["id"] == elem:
                    params["os"][i] = platform["name"]

        # Creating the parameters for the sql query
        sql_params = {
            "from_date": params["from_date"],
            "to_date": params["to_date"],
            "limit": params["result_number"],
            "offset": params["result_offset"]
        }
        sql_params = Search.dispatch_params(sql_params, "term",
                                            params["terms"])
        sql_params = Search.dispatch_params(sql_params, "product",
                                            params["products"])
        sql_params = Search.dispatch_params(sql_params, "os",
                                            params["os"])
        sql_params = Search.dispatch_params(sql_params, "version",
                                            params["versions"])
        sql_params = Search.dispatch_params(sql_params, "build",
                                            params["build_ids"])
        sql_params = Search.dispatch_params(sql_params, "reason",
                                            params["reasons"])
        sql_params = Search.dispatch_params(sql_params, "plugin_term",
                                            params["plugin_terms"])
        sql_params = Search.dispatch_params(sql_params, "branch",
                                            params["branches"])

        # Preparing the different parts of the sql query

        #---------------------------------------------------------------
        # SELECT
        #---------------------------------------------------------------

        sql_select = self.generate_sql_select(params)

        # Adding count for each OS
        for i in self.context.platforms:
            sql_params["os_%s" % i["id"]] = i["name"]

        #---------------------------------------------------------------
        # FROM
        #---------------------------------------------------------------

        sql_from = self.generate_sql_from(params)

        #---------------------------------------------------------------
        # WHERE
        #---------------------------------------------------------------

        sql_where = ["""
            WHERE r.date_processed BETWEEN %(from_date)s AND %(to_date)s
        """]

        ## Adding terms to where clause
        if params["terms"]:
            if params["search_mode"] == "is_exactly":
                sql_where.append("r.signature=%(term)s")
#.........这里部分代码省略.........

    def test_build_query_from_params(self):
        # Test with all default parameters
        config = self.get_dummy_context()
        params = {}
        params = scommon.get_parameters(params)
        query = ElasticSearchBase.build_query_from_params(params, config)
        self.assertTrue(query)
        self.assertTrue("query" in query)
        self.assertTrue("size" in query)
        self.assertTrue("from" in query)

        # Searching for a term in a specific field and with a specific product
        params = {
            "terms": "hang",
            "fields": "dump",
            "search_mode": "contains",
            "products": "fennec"
        }
        params = scommon.get_parameters(params)
        query = ElasticSearchBase.build_query_from_params(params, config)
        self.assertTrue(query)
        self.assertTrue("query" in query)
        self.assertTrue("filtered" in query["query"])

        filtered = query["query"]["filtered"]
        self.assertTrue("query" in filtered)
        self.assertTrue("wildcard" in filtered["query"])
        self.assertTrue("dump" in filtered["query"]["wildcard"])

        dump_term = filtered["query"]["wildcard"]["dump"]
        self.assertEqual(dump_term, "*hang*")
        self.assertTrue("filter" in filtered)
        self.assertTrue("and" in filtered["filter"])

        # Test versions
        params = {
            "products": "WaterWolf",
            "versions": "WaterWolf:1.0a1"
        }
        params = scommon.get_parameters(params)
        params['versions_info'] = {
            'WaterWolf:1.0a1': {
                "version_string": "1.0a1",
                "product_name": "WaterWolf",
                "major_version": "1.0a1",
                "release_channel": "nightly-water",
                "build_id": None
            }
        }
        query = ElasticSearchBase.build_query_from_params(params, config)
        filtered = query["query"]["filtered"]

        self.assertTrue("and" in filtered["filter"])
        and_filter_str = json.dumps(filtered["filter"]['and'])
        self.assertTrue('WaterWolf' in and_filter_str)
        self.assertTrue('1.0a1' in and_filter_str)
        self.assertTrue('nightly-water' in and_filter_str)

        # Test versions with an empty release channel in versions_info
        params = {
            "products": "WaterWolf",
            "versions": "WaterWolf:2.0"
        }
        params = scommon.get_parameters(params)
        params['versions_info'] = {
            'WaterWolf:2.0': {
                "version_string": "2.0",
                "product_name": "WaterWolf",
                "major_version": "2.0",
                "release_channel": None,
                "build_id": None
            }
        }
        query = ElasticSearchBase.build_query_from_params(params, config)
        filtered = query["query"]["filtered"]

        self.assertTrue("and" in filtered["filter"])
        and_filter_str = json.dumps(filtered["filter"]['and'])
        self.assertTrue('WaterWolf' in and_filter_str)
        self.assertTrue('2.0' in and_filter_str)

    def get_list(self, **kwargs):
        """
        List all crashes with a given signature and return them.

        Optional arguments: see SearchCommon.get_parameters()

        """
        # aliases
        if "from" in kwargs and "from_date" not in kwargs:
            kwargs["from_date"] = kwargs.get("from")
        if "to" in kwargs and "to_date" not in kwargs:
            kwargs["to_date"] = kwargs.get("to")

        params = search_common.get_parameters(kwargs)

        if not params["signature"]:
            raise MissingOrBadArgumentError("Mandatory parameter 'signature' is missing or empty")

        params["terms"] = params["signature"]
        params["search_mode"] = "is_exactly"

        # Default mode falls back to starts_with for postgres
        if params["plugin_search_mode"] == "default":
            params["plugin_search_mode"] = "starts_with"

        # Limiting to a signature
        if params["terms"]:
            params["terms"] = self.prepare_terms(params["terms"], params["search_mode"])

        # Searching for terms in plugins
        if params["report_process"] == "plugin" and params["plugin_terms"]:
            params["plugin_terms"] = " ".join(params["plugin_terms"])
            params["plugin_terms"] = self.prepare_terms(params["plugin_terms"], params["plugin_search_mode"])

        # Get information about the versions
        util_service = Util(config=self.context)
        params["versions_info"] = util_service.versions_info(**params)

        # Parsing the versions
        params["versions_string"] = params["versions"]
        (params["versions"], params["products"]) = self.parse_versions(params["versions"], params["products"])

        if hasattr(self.context, "webapi"):
            context = self.context.webapi
        else:
            # old middleware
            context = self.context
        # Changing the OS ids to OS names
        for i, elem in enumerate(params["os"]):
            for platform in context.platforms:
                if platform["id"][:3] == elem[:3]:
                    params["os"][i] = platform["name"]

        # Creating the parameters for the sql query
        sql_params = {}

        # Preparing the different parts of the sql query

        sql_select = """
            SELECT
                r.date_processed,
                r.uptime,
                r.user_comments,
                r.uuid,
                r.product,
                r.version,
                r.build,
                r.signature,
                r.url,
                r.os_name,
                r.os_version,
                r.cpu_name,
                r.cpu_info,
                r.address,
                r.reason,
                r.last_crash,
                r.install_age,
                r.hangid,
                r.process_type,
                (r.client_crash_date - (r.install_age * INTERVAL '1 second'))
                    AS install_time,
                rd.duplicate_of
        """

        sql_from = self.build_reports_sql_from(params)
        sql_from = (
            """%s
            LEFT OUTER JOIN reports_duplicates rd ON r.uuid = rd.uuid
        """
            % sql_from
        )

        (sql_where, sql_params) = self.build_reports_sql_where(params, sql_params, self.context)

        sql_order = """
            ORDER BY r.date_processed DESC
        """

        (sql_limit, sql_params) = self.build_reports_sql_limit(params, sql_params)

#.........这里部分代码省略.........

    def get(self, **kwargs):
        """
        Search for crashes and return them.

        See http://socorro.readthedocs.org/en/latest/middleware.html#search

        Optional arguments: see SearchCommon.get_parameters()

        """
        # change aliases from the web to the implementation's need
        if "for" in kwargs and "terms" not in kwargs:
            kwargs["terms"] = kwargs.get("for")
        if "from" in kwargs and "from_date" not in kwargs:
            kwargs["from_date"] = kwargs.get("from")
        if "to" in kwargs and "to_date" not in kwargs:
            kwargs["to_date"] = kwargs.get("to")
        if "in" in kwargs and "fields" not in kwargs:
            kwargs["fields"]