Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use statistics method in pandera
Best Python code snippet using pandera_python
test_statscompiler.py
Source:test_statscompiler.py 
...18import multiprocessing19import pytest20from unittest.mock import call21from supvisors.statscompiler import *22def test_cpu_statistics():23 """ Test the CPU statistics between 2 dates. """24 # take 2 spaced instant cpu statistics25 ref_stats = [(83.31, 305.4)] * (multiprocessing.cpu_count() + 1)26 last_stats = [(83.32, 306.4)] * (multiprocessing.cpu_count() + 1)27 stats = cpu_statistics(last_stats, ref_stats)28 # test number of results (number of cores + average)29 assert len(stats) == multiprocessing.cpu_count() + 130 # test bounds (percent)31 for cpu in stats:32 assert type(cpu) is float33 assert cpu >= 034 assert cpu <= 10035def test_cpu_total_work():36 """ Test the CPU total work between 2 dates. """37 # take 2 instant cpu statistics38 ref_stats = [(83.31, 305.4)] * 239 last_stats = [(83.41, 306.3)] * 240 total_work = cpu_total_work(last_stats, ref_stats)41 # total work is the sum of jiffies spent on cpu all42 assert pytest.approx(total_work) == 143def test_io_statistics():44 """ Test the I/O statistics between 2 dates. """45 # take 2 instant cpu statistics46 ref_stats = {'eth0': (2000, 200), 'lo': (5000, 5000)}47 last_stats = {'eth0': (2896, 328), 'lo': (6024, 6024)}48 stats = io_statistics(last_stats, ref_stats, 1)49 # test keys50 assert stats.keys() == ref_stats.keys()51 assert stats.keys() == last_stats.keys()52 # test that values53 assert stats == {'lo': (8, 8), 'eth0': (7, 1)}54def test_cpu_process_statistics():55 """ Test the CPU of the process between 2 dates. """56 stats = cpu_process_statistics(50, 20, 100)57 assert type(stats) is float58 assert pytest.approx(stats) == 3059def test_statistics():60 """ Test the global statistics between 2 dates. """61 ref_stats = (1000, [(25, 400), (25, 125), (15, 150)], 65, {'eth0': (2000, 200), 'lo': (5000, 5000)},62 {'myself': (26088, (0.15, 1.85))})63 last_stats = (1002, [(45, 700), (50, 225), (40, 250)], 67.7, {'eth0': (2768, 456), 'lo': (6024, 6024)},64 {'myself': (26088, (1.75, 1.9))})65 stats = statistics(last_stats, ref_stats)66 # check result67 assert len(stats) == 568 date, cpu_stats, mem_stats, io_stats, proc_stats = stats69 # check date70 assert date == 100271 # check cpu72 assert cpu_stats == [6.25, 20.0, 20.0]73 # check memory74 assert pytest.approx(mem_stats) == 67.775 # check io76 assert io_stats == {'lo': (4, 4), 'eth0': (3, 1)}77 # check process stats78 assert proc_stats == {('myself', 26088): (0.5, 1.9)}79@pytest.fixture80def statistics_instance(supvisors):81 # testing with period 12 and history depth 282 return StatisticsInstance(12, 2, supvisors.logger)83def test_stats_create(statistics_instance):84 """ Test the initialization of an instance. """85 # check attributes86 assert statistics_instance.period == 287 assert statistics_instance.depth == 288 assert statistics_instance.counter == -189 assert statistics_instance.ref_stats is None90 assert type(statistics_instance.cpu) is list91 assert not statistics_instance.cpu92 assert type(statistics_instance.mem) is list93 assert not statistics_instance.mem94 assert type(statistics_instance.io) is dict95 assert not statistics_instance.io96 assert type(statistics_instance.proc) is dict97 assert not statistics_instance.proc98def test_clear(statistics_instance):99 """ Test the clearance of an instance. """100 # change values101 statistics_instance.counter = 28102 statistics_instance.ref_stats = ('dummy', 0)103 statistics_instance.cpu = [13.2, 14.8]104 statistics_instance.mem = [56.4, 71.3, 68.9]105 statistics_instance.io = {'eth0': (123465, 654321), 'lo': (321, 321)}106 statistics_instance.proc = {('myself', 5888): (25.0, 12.5)}107 # check clearance108 statistics_instance.clear()109 assert statistics_instance.period == 2110 assert statistics_instance.depth == 2111 assert statistics_instance.counter == -1112 assert statistics_instance.ref_stats is None113 assert type(statistics_instance.cpu) is list114 assert not statistics_instance.cpu115 assert type(statistics_instance.mem) is list116 assert not statistics_instance.mem117 assert type(statistics_instance.io) is dict118 assert not statistics_instance.io119 assert type(statistics_instance.proc) is dict120 assert not statistics_instance.proc121def test_find_process_stats(statistics_instance):122 """ Test the search method for process statistics. """123 # change values124 statistics_instance.proc = {('the_other', 1234): (1.5, 2.4), ('myself', 5888): (25.0, 12.5)}125 # test find method with wrong argument126 assert statistics_instance.find_process_stats('someone') is None127 # test find method with correct argument128 stats = statistics_instance.find_process_stats('myself')129 assert stats == (25.0, 12.5)130def test_trunc_depth(statistics_instance):131 """ Test the history depth. """132 # test that the trunc_depth method does nothing when less than 2 elements in list133 test_list = [1, 2]134 statistics_instance.trunc_depth(test_list)135 assert test_list == [1, 2]136 # test that the trunc_depth method keeps only the last 5 elements in list137 test_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]138 statistics_instance.trunc_depth(test_list)139 assert test_list == [9, 10]140def test_push_cpu_stats(statistics_instance):141 """ Test the storage of the CPU instant statistics. """142 cpu_stats = [12.1, 16.0]143 # test adding CPU stats on non-initialized structure144 # should not happen as protected by the upper call of push_statistics145 statistics_instance._push_cpu_stats(cpu_stats)146 assert not statistics_instance.cpu147 # init internal structure and retry148 statistics_instance.cpu = [[] for _ in cpu_stats]149 statistics_instance._push_cpu_stats(cpu_stats)150 assert statistics_instance.cpu == [[12.1], [16.0]]151 # again: list increases152 cpu_stats = [8.7, 14.6]153 statistics_instance._push_cpu_stats(cpu_stats)154 assert statistics_instance.cpu == [[12.1, 8.7], [16.0, 14.6]]155 # again: list rotates due to history depth at 2156 cpu_stats = [11.9, 5.5]157 statistics_instance._push_cpu_stats(cpu_stats)158 assert statistics_instance.cpu == [[8.7, 11.9], [14.6, 5.5]]159def test_push_mem_stats(statistics_instance):160 """ Test the storage of the MEM instant statistics. """161 # MEM stats internal structure is initialized (simple list)162 statistics_instance._push_mem_stats(12.1)163 assert statistics_instance.mem == [12.1]164 # again: list increases165 statistics_instance._push_mem_stats(8.7)166 assert statistics_instance.mem == [12.1, 8.7]167 # again: list rotates due to history depth at 2168 statistics_instance._push_mem_stats(11.9)169 assert statistics_instance.mem == [8.7, 11.9]170def test_push_io_stats(statistics_instance):171 """ Test the storage of the IO instant statistics. """172 io_stats = {'eth0': (1024, 2000), 'lo': (500, 500)}173 # first time: structures are created174 statistics_instance._push_io_stats(io_stats)175 assert statistics_instance.io == {'eth0': ([1024], [2000]), 'lo': ([500], [500])}176 # again: list increases177 io_stats = {'eth0': (1250, 2200), 'lo': (620, 620)}178 statistics_instance._push_io_stats(io_stats)179 assert statistics_instance.io == {'eth0': ([1024, 1250], [2000, 2200]), 'lo': ([500, 620], [500, 620])}180 # again: list rotates due to history depth at 2181 io_stats = {'eth0': (2048, 2512), 'lo': (756, 756)}182 statistics_instance._push_io_stats(io_stats)183 assert statistics_instance.io == {'eth0': ([1250, 2048], [2200, 2512]), 'lo': ([620, 756], [620, 756])}184 # test obsolete and new interface185 io_stats = {'eth1': (3072, 2768), 'lo': (1780, 1780)}186 statistics_instance._push_io_stats(io_stats)187 assert statistics_instance.io == {'eth1': ([3072], [2768]), 'lo': ([756, 1780], [756, 1780])}188def test_push_process_stats(statistics_instance):189 """ Test the storage of the process instant statistics. """190 proc_stats = {('myself', 118612): (0.15, 1.85), ('other', 7754): (15.4, 12.2)}191 statistics_instance._push_process_stats(proc_stats)192 assert statistics_instance.proc == {('myself', 118612): ([0.15], [1.85]), ('other', 7754): ([15.4], [12.2])}193 # again: list increases194 proc_stats = {('myself', 118612): (1.9, 1.93), ('other', 7754): (14.9, 12.8)}195 statistics_instance._push_process_stats(proc_stats)196 assert statistics_instance.proc == {('myself', 118612): ([0.15, 1.9], [1.85, 1.93]),197 ('other', 7754): ([15.4, 14.9], [12.2, 12.8])}198 # again: list rotates due to history depth at 2199 proc_stats = {('myself', 118612): (2.47, 2.04), ('other', 7754): (6.5, 13.0)}200 statistics_instance._push_process_stats(proc_stats)201 assert statistics_instance.proc == {('myself', 118612): ([1.9, 2.47], [1.93, 2.04]),202 ('other', 7754): ([14.9, 6.5], [12.8, 13.0])}203 # test obsolete and new processes (here other is just restarted - new pid)204 proc_stats = {('myself', 118612): (0.15, 1.85), ('other', 826): (1.89, 2.67)}205 statistics_instance._push_process_stats(proc_stats)206 assert statistics_instance.proc == {('myself', 118612): ([2.47, 0.15], [2.04, 1.85]),207 ('other', 826): ([1.89], [2.67])}208def test_push_statistics(mocker, statistics_instance):209 """ Test the storage of the instant statistics. """210 mocked_stats = mocker.patch('supvisors.statscompiler.statistics')211 mocked_cpu = mocker.patch.object(statistics_instance, '_push_cpu_stats')212 mocked_mem = mocker.patch.object(statistics_instance, '_push_mem_stats')213 mocked_io = mocker.patch.object(statistics_instance, '_push_io_stats')214 mocked_proc = mocker.patch.object(statistics_instance, '_push_process_stats')215 # push first set of measures216 stats1 = 'stats 1'217 mocked_stats.return_value = (8.5, 'cpu_stats 1', 76.1, 'io_stats 1', 'proc_stats 1')218 statistics_instance.push_statistics(stats1)219 # check evolution of instance220 assert statistics_instance.counter == 0221 assert statistics_instance.ref_stats is stats1222 assert not mocked_stats.called223 assert not mocked_cpu.called224 assert not mocked_mem.called225 assert not mocked_io.called226 assert not mocked_proc.called227 # push second set of measures228 stats2 = 'stats 2'229 mocked_stats.return_value = (18.52, 'cpu_stats 2', 76.2, 'io_stats 2', 'proc_stats 2')230 statistics_instance.push_statistics(stats2)231 # counter is based a theoretical period of 5 seconds so this update is not taken into account232 # check evolution of instance233 assert statistics_instance.counter == 1234 assert statistics_instance.ref_stats is stats1235 assert not mocked_stats.called236 assert not mocked_cpu.called237 assert not mocked_mem.called238 assert not mocked_io.called239 assert not mocked_proc.called240 # push third set of measures241 stats3 = 'stats 3'242 mocked_stats.return_value = (28.5, 'cpu_stats 3', 76.1, 'io_stats 3', 'proc_stats 3')243 statistics_instance.push_statistics(stats3)244 # this update is taken into account245 # check evolution of instance246 assert statistics_instance.counter == 2247 assert statistics_instance.ref_stats is stats3248 assert mocked_stats.call_args_list == [call(stats3, stats1)]249 assert mocked_cpu.call_args_list == [call('cpu_stats 3')]250 assert mocked_mem.call_args_list == [call(76.1)]251 assert mocked_io.call_args_list == [call('io_stats 3')]252 assert mocked_proc.call_args_list == [call('proc_stats 3')]253 mocker.resetall()254 # push fourth set of measures (reuse stats2)255 statistics_instance.push_statistics(stats2)256 # again,this update is not taken into account257 assert statistics_instance.counter == 3258 assert statistics_instance.ref_stats is stats3259 assert not mocked_stats.called260 assert not mocked_cpu.called261 assert not mocked_mem.called262 assert not mocked_io.called263 assert not mocked_proc.called264 # push fifth set of measures265 stats5 = 'stats 5'266 mocked_stats.return_value = (38.5, 'cpu_stats 5', 75.9, 'io_stats 5', 'proc_stats 5')267 statistics_instance.push_statistics(stats5)268 # this update is taken into account269 # check evolution of instance270 assert statistics_instance.counter == 4271 assert statistics_instance.ref_stats is stats5272 assert mocked_stats.call_args_list == [call(stats5, stats3)]273 assert mocked_cpu.call_args_list == [call('cpu_stats 5')]274 assert mocked_mem.call_args_list == [call(75.9)]275 assert mocked_io.call_args_list == [call('io_stats 5')]276 assert mocked_proc.call_args_list == [call('proc_stats 5')]277@pytest.fixture278def compiler(supvisors):279 return StatisticsCompiler(supvisors)280def test_compiler_create(supvisors, compiler):281 """ Test the initialization for statistics of all addresses. """282 # check compiler contents at initialisation283 assert list(compiler.data.keys()) == list(supvisors.supvisors_mapper.instances.keys())284 for period_instance in compiler.data.values():285 assert tuple(period_instance.keys()) == supvisors.options.stats_periods286 for period, instance in period_instance.items():287 assert type(instance) is StatisticsInstance288 assert instance.period == period / 5289 assert instance.depth == supvisors.options.stats_histo290def test_compiler_clear(compiler):291 """ Test the clearance for statistics of all addresses. """292 # set data to a given address293 for address, period_instance in compiler.data.items():294 for period, instance in period_instance.items():295 instance.counter = 28296 instance.ref_stats = ('dummy', 0)297 instance.cpu = [13.2, 14.8]298 instance.mem = [56.4, 71.3, 68.9]299 instance.io = {'eth0': (123465, 654321), 'lo': (321, 321)}300 instance.proc = {('myself', 5888): (25.0, 12.5)}301 # check clearance of instance302 compiler.clear('10.0.0.2')303 for address, period_instance in compiler.data.items():304 if address == '10.0.0.2':305 for period, instance in period_instance.items():306 assert instance.period == period / 5307 assert instance.depth == 10308 assert instance.counter == -1309 assert instance.ref_stats is None310 assert type(instance.cpu) is list311 assert not instance.cpu312 assert type(instance.mem) is list313 assert not instance.mem314 assert type(instance.io) is dict315 assert not instance.io316 assert type(instance.proc) is dict317 assert not instance.proc318 else:319 for period, instance in period_instance.items():320 assert instance.period == period / 5321 assert instance.depth == 10322 assert instance.counter == 28323 assert instance.ref_stats == ('dummy', 0)324 assert instance.cpu == [13.2, 14.8]325 assert instance.mem == [56.4, 71.3, 68.9]326 assert instance.io == {'eth0': (123465, 654321), 'lo': (321, 321)}327 assert instance.proc == {('myself', 5888): (25.0, 12.5)}328def test_compiler_push_statistics(compiler):329 """ Test the storage of the instant statistics of an address. """330 # push statistics to a given address331 stats1 = (8.5, [(25, 400), (25, 125), (15, 150), (40, 400), (20, 200)],332 76.1, {'eth0': (1024, 2000), 'lo': (500, 500)}, {'myself': (118612, (0.15, 1.85))})333 compiler.push_statistics('10.0.0.2', stats1)334 # check compiler contents335 for address, period_instance in compiler.data.items():336 if address == '10.0.0.2':337 for period, instance in period_instance.items():338 assert instance.counter == 0339 assert instance.ref_stats is stats1340 else:341 for period, instance in period_instance.items():342 assert instance.counter == -1343 assert instance.ref_stats is None344 # push statistics to a given address345 stats2 = (28.5, [(45, 700), (50, 225), (40, 250), (42, 598), (20, 400)],346 76.1, {'eth0': (2048, 2512), 'lo': (756, 756)}, {'myself': (118612, (1.75, 1.9))})347 compiler.push_statistics('10.0.0.2', stats2)348 # check compiler contents349 for address, period_instance in compiler.data.items():350 if address == '10.0.0.2':351 for period, instance in period_instance.items():352 assert instance.counter == 1353 if period == 5:354 assert instance.ref_stats is stats2355 else:356 assert instance.ref_stats is stats1357 else:358 for period, instance in period_instance.items():359 assert instance.counter == -1360 assert instance.ref_stats is None361 # push statistics to a given address362 stats3 = (38.5, [(80, 985), (89, 386), (48, 292), (42, 635), (32, 468)],363 75.9, {'eth0': (3072, 2768), 'lo': (1780, 1780)}, {'myself': (118612, (11.75, 1.87))})364 compiler.push_statistics('10.0.0.2', stats3)365 # check compiler contents366 for address, period_instance in compiler.data.items():367 if address == '10.0.0.2':368 for period, instance in period_instance.items():369 assert instance.counter == 2370 if period == 5:371 assert instance.ref_stats is stats3372 else:373 assert instance.ref_stats is stats1374 else:375 for period, instance in period_instance.items():376 assert instance.counter == -1377 assert instance.ref_stats is None378 # push statistics to a given address379 stats4 = (48.5, [(84, 1061), (92, 413), (48, 480), (45, 832), (40, 1100)],380 74.7, {'eth0': (3584, 3792), 'lo': (1812, 1812)}, {'myself': (118612, (40.75, 2.34))})381 compiler.push_statistics('10.0.0.2', stats4)382 # check compiler contents383 for address, period_instance in compiler.data.items():384 if address == '10.0.0.2':385 for period, instance in period_instance.items():386 assert instance.counter == 3387 if period in [5, 15]:388 assert instance.ref_stats is stats4389 else:390 assert instance.ref_stats is stats1391 else:392 for period, instance in period_instance.items():393 assert instance.counter == -1...
test_issues.py
Source:test_issues.py
...32 status=200,33 )34 yield rsps35@pytest.fixture36def resp_issue_statistics():37 content = {"statistics": {"counts": {"all": 20, "closed": 5, "opened": 15}}}38 with responses.RequestsMock() as rsps:39 rsps.add(40 method=responses.GET,41 url=re.compile(42 r"http://localhost/api/v4/((groups|projects)/1/)?issues_statistics"43 ),44 json=content,45 content_type="application/json",46 status=200,47 )48 yield rsps49def test_list_issues(gl, resp_list_issues):50 data = gl.issues.list()51 assert data[1].id == 252 assert data[1].name == "other_name"53def test_get_issue(gl, resp_get_issue):54 issue = gl.issues.get(1)55 assert issue.id == 156 assert issue.name == "name"57def test_get_issues_statistics(gl, resp_issue_statistics):58 statistics = gl.issues_statistics.get()59 assert isinstance(statistics, IssuesStatistics)60 assert statistics.statistics["counts"]["all"] == 2061def test_get_group_issues_statistics(group, resp_issue_statistics):62 statistics = group.issues_statistics.get()63 assert isinstance(statistics, GroupIssuesStatistics)64 assert statistics.statistics["counts"]["all"] == 2065def test_get_project_issues_statistics(project, resp_issue_statistics):66 statistics = project.issues_statistics.get()67 assert isinstance(statistics, ProjectIssuesStatistics)68 assert statistics.statistics["counts"]["all"] == 2069 # Deprecated attribute70 deprecated = project.issuesstatistics.get()...
ticky_check.py
Source:ticky_check.py
1#!/usr/bin/env python32import sys3import os4import re5import operator6import csv7log_file = "syslog.log"8error_message_csv = "error_message.csv"9error_message_html = "/var/www/html/error_message.html"10user_statistics_csv = "user_statistics.csv"11user_statistics_html = "/var/www/html/user_statistics.html"12error_messages = {}13user_statistics = {}14with open(log_file) as log:15 while True:16 # print("Reading log entry")17 log_entry = log.readline().strip()18 if not log_entry:19 # print("Last record read")20 break21 parsed_entry = re.search(r"^[\w :.]+ticky: ((?:INFO)|(?:ERROR)) (\w[\w ']*)[^(]*\(([\w.]*)\)$", log_entry)22 # print(parsed_entry)23 if parsed_entry == None:24 # print("Valid log entry not found. Getting next record")25 continue26 # print(parsed_entry)27 if parsed_entry.group(1).strip() == "ERROR":28 # print("ERROR entry found")29 if parsed_entry.group(2).strip() not in error_messages:30 error_messages[parsed_entry.group(2).strip()] = 031 error_messages[parsed_entry.group(2).strip()] += 132 if parsed_entry.group(3) not in user_statistics:33 user_statistics[parsed_entry.group(3).strip()] = [0,0]34 user_statistics[parsed_entry.group(3).strip()][1] += 135 continue36 if parsed_entry.group(1).strip() == "INFO":37 # print("INFO entry found")38 if parsed_entry.group(3).strip() not in user_statistics:39 user_statistics[parsed_entry.group(3).strip()] = [0,0]40 user_statistics[parsed_entry.group(3).strip()][0] += 141error_messages_sorted = sorted(error_messages.items(), key=operator.itemgetter(1), reverse=True)42user_statistics_sorted = [(row[0],row[1][0],row[1][1])for row in sorted(user_statistics.items())]43# print(user_statistics_sorted)44error_messages_sorted.insert(0, ("Error", "Count"))45# print(error_messages_sorted)46user_statistics_sorted.insert(0, ("Username", "INFO", "ERROR"))47# print(user_statistics_sorted)48with open(error_message_csv, "w", newline = "") as error_message_csv_file:49 error_writer = csv.writer(error_message_csv_file)50 error_writer.writerows(error_messages_sorted)51with open(user_statistics_csv, "w", newline="") as user_statistics_csv_file:52 user_writer = csv.writer(user_statistics_csv_file)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pandera automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
