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