xref: /aosp_15_r20/external/perfetto/python/test/bigtrace_gcs_manualtest.py (revision 6dbdd20afdafa5e3ca9b8809fa73465d530080dc)
1# Copyright (C) 2024 The Android Open Source Project
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import os
16import perfetto.bigtrace.api
17import subprocess
18import unittest
19
20from perfetto.common.exceptions import PerfettoException
21
22# To run this test you must setup the a GCS bucket and a GKE cluster setup with
23# Bigtrace running.
24# This should be executed within the same VPC to allow for connection to the
25# service.
26
27# This should be replaced with the name of the trace bucket you have deployed
28# on.
29TRACE_BUCKET_NAME = "trace_example_bucket"
30# This should be loaded in the top level of the bucket.
31TRACE_PATH = "android_startup_real.perfetto_trace"
32# This should be replaced with the address of the Orchestrator service for the
33# Bigtrace service.
34ORCHESTRATOR_ADDRESS = "127.0.0.1:5052"
35# This can be changed if testing on a different trace.
36QUERY_RESULT_COUNT = 339338
37
38
39class BigtraceGcsTest(unittest.TestCase):
40
41  def setUpClass(self):
42    self.client = perfetto.bigtrace.api.Bigtrace(
43        wait_for_ready_for_testing=True)
44
45  def test_valid_trace(self):
46    traces = [f"/gcs/{TRACE_BUCKET_NAME}/o/{TRACE_PATH}"]
47    result = self.client.query(traces, "SELECT count(1) as count FROM slice")
48    self.assertEqual(result['count'].iloc[0], QUERY_RESULT_COUNT)
49
50  def test_invalid_trace(self):
51    with self.assertRaises(PerfettoException):
52      traces = [f"/gcs/{TRACE_BUCKET_NAME}/o/badpath"]
53      result = self.client.query(traces, "SELECT count(1) as count FROM slice")
54
55  def test_invalid_bucket(self):
56    with self.assertRaises(PerfettoException):
57      traces = [f"/gcs//o/{TRACE_PATH}"]
58      result = self.client.query(traces, "SELECT count(1) as count FROM slice")
59