KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use stat method in lisa
Best Python code snippet using lisa_python
test_unit_externals_status.py
Source:test_unit_externals_status.py 
1#!/usr/bin/env python32"""Unit test driver for the manic external status reporting module.3Note: this script assumes the path to the manic package is already in4the python path.5"""6from __future__ import absolute_import7from __future__ import unicode_literals8from __future__ import print_function9import unittest10from manic.externals_status import ExternalStatus11class TestStatusObject(unittest.TestCase):12 """Verify that the Status object behaives as expected.13 """14 def test_exists_empty_all(self):15 """If the repository sync-state is empty (doesn't exist), and there is no16 clean state, then it is considered not to exist.17 """18 stat = ExternalStatus()19 stat.sync_state = ExternalStatus.EMPTY20 stat.clean_state = ExternalStatus.DEFAULT21 exists = stat.exists()22 self.assertFalse(exists)23 stat.clean_state = ExternalStatus.EMPTY24 exists = stat.exists()25 self.assertFalse(exists)26 stat.clean_state = ExternalStatus.UNKNOWN27 exists = stat.exists()28 self.assertFalse(exists)29 # this state represtens an internal logic error in how the30 # repo status was determined.31 stat.clean_state = ExternalStatus.STATUS_OK32 exists = stat.exists()33 self.assertTrue(exists)34 # this state represtens an internal logic error in how the35 # repo status was determined.36 stat.clean_state = ExternalStatus.DIRTY37 exists = stat.exists()38 self.assertTrue(exists)39 def test_exists_default_all(self):40 """If the repository sync-state is default, then it is considered to exist41 regardless of clean state.42 """43 stat = ExternalStatus()44 stat.sync_state = ExternalStatus.DEFAULT45 stat.clean_state = ExternalStatus.DEFAULT46 exists = stat.exists()47 self.assertTrue(exists)48 stat.clean_state = ExternalStatus.EMPTY49 exists = stat.exists()50 self.assertTrue(exists)51 stat.clean_state = ExternalStatus.UNKNOWN52 exists = stat.exists()53 self.assertTrue(exists)54 stat.clean_state = ExternalStatus.STATUS_OK55 exists = stat.exists()56 self.assertTrue(exists)57 stat.clean_state = ExternalStatus.DIRTY58 exists = stat.exists()59 self.assertTrue(exists)60 def test_exists_unknown_all(self):61 """If the repository sync-state is unknown, then it is considered to exist62 regardless of clean state.63 """64 stat = ExternalStatus()65 stat.sync_state = ExternalStatus.UNKNOWN66 stat.clean_state = ExternalStatus.DEFAULT67 exists = stat.exists()68 self.assertTrue(exists)69 stat.clean_state = ExternalStatus.EMPTY70 exists = stat.exists()71 self.assertTrue(exists)72 stat.clean_state = ExternalStatus.UNKNOWN73 exists = stat.exists()74 self.assertTrue(exists)75 stat.clean_state = ExternalStatus.STATUS_OK76 exists = stat.exists()77 self.assertTrue(exists)78 stat.clean_state = ExternalStatus.DIRTY79 exists = stat.exists()80 self.assertTrue(exists)81 def test_exists_modified_all(self):82 """If the repository sync-state is modified, then it is considered to exist83 regardless of clean state.84 """85 stat = ExternalStatus()86 stat.sync_state = ExternalStatus.MODEL_MODIFIED87 stat.clean_state = ExternalStatus.DEFAULT88 exists = stat.exists()89 self.assertTrue(exists)90 stat.clean_state = ExternalStatus.EMPTY91 exists = stat.exists()92 self.assertTrue(exists)93 stat.clean_state = ExternalStatus.UNKNOWN94 exists = stat.exists()95 self.assertTrue(exists)96 stat.clean_state = ExternalStatus.STATUS_OK97 exists = stat.exists()98 self.assertTrue(exists)99 stat.clean_state = ExternalStatus.DIRTY100 exists = stat.exists()101 self.assertTrue(exists)102 def test_exists_ok_all(self):103 """If the repository sync-state is ok, then it is considered to exist104 regardless of clean state.105 """106 stat = ExternalStatus()107 stat.sync_state = ExternalStatus.STATUS_OK108 stat.clean_state = ExternalStatus.DEFAULT109 exists = stat.exists()110 self.assertTrue(exists)111 stat.clean_state = ExternalStatus.EMPTY112 exists = stat.exists()113 self.assertTrue(exists)114 stat.clean_state = ExternalStatus.UNKNOWN115 exists = stat.exists()116 self.assertTrue(exists)117 stat.clean_state = ExternalStatus.STATUS_OK118 exists = stat.exists()119 self.assertTrue(exists)120 stat.clean_state = ExternalStatus.DIRTY121 exists = stat.exists()122 self.assertTrue(exists)123 def test_update_ok_all(self):124 """If the repository in-sync is ok, then it is safe to125 update only if clean state is ok126 """127 stat = ExternalStatus()128 stat.sync_state = ExternalStatus.STATUS_OK129 stat.clean_state = ExternalStatus.DEFAULT130 safe_to_update = stat.safe_to_update()131 self.assertFalse(safe_to_update)132 stat.clean_state = ExternalStatus.EMPTY133 safe_to_update = stat.safe_to_update()134 self.assertFalse(safe_to_update)135 stat.clean_state = ExternalStatus.UNKNOWN136 safe_to_update = stat.safe_to_update()137 self.assertFalse(safe_to_update)138 stat.clean_state = ExternalStatus.STATUS_OK139 safe_to_update = stat.safe_to_update()140 self.assertTrue(safe_to_update)141 stat.clean_state = ExternalStatus.DIRTY142 safe_to_update = stat.safe_to_update()143 self.assertFalse(safe_to_update)144 def test_update_modified_all(self):145 """If the repository in-sync is modified, then it is safe to146 update only if clean state is ok147 """148 stat = ExternalStatus()149 stat.sync_state = ExternalStatus.MODEL_MODIFIED150 stat.clean_state = ExternalStatus.DEFAULT151 safe_to_update = stat.safe_to_update()152 self.assertFalse(safe_to_update)153 stat.clean_state = ExternalStatus.EMPTY154 safe_to_update = stat.safe_to_update()155 self.assertFalse(safe_to_update)156 stat.clean_state = ExternalStatus.UNKNOWN157 safe_to_update = stat.safe_to_update()158 self.assertFalse(safe_to_update)159 stat.clean_state = ExternalStatus.STATUS_OK160 safe_to_update = stat.safe_to_update()161 self.assertTrue(safe_to_update)162 stat.clean_state = ExternalStatus.DIRTY163 safe_to_update = stat.safe_to_update()164 self.assertFalse(safe_to_update)165 def test_update_unknown_all(self):166 """If the repository in-sync is unknown, then it is not safe to167 update, regardless of the clean state.168 """169 stat = ExternalStatus()170 stat.sync_state = ExternalStatus.UNKNOWN171 stat.clean_state = ExternalStatus.DEFAULT172 safe_to_update = stat.safe_to_update()173 self.assertFalse(safe_to_update)174 stat.clean_state = ExternalStatus.EMPTY175 safe_to_update = stat.safe_to_update()176 self.assertFalse(safe_to_update)177 stat.clean_state = ExternalStatus.UNKNOWN178 safe_to_update = stat.safe_to_update()179 self.assertFalse(safe_to_update)180 stat.clean_state = ExternalStatus.STATUS_OK181 safe_to_update = stat.safe_to_update()182 self.assertFalse(safe_to_update)183 stat.clean_state = ExternalStatus.DIRTY184 safe_to_update = stat.safe_to_update()185 self.assertFalse(safe_to_update)186 def test_update_default_all(self):187 """If the repository in-sync is default, then it is not safe to188 update, regardless of the clean state.189 """190 stat = ExternalStatus()191 stat.sync_state = ExternalStatus.UNKNOWN192 stat.clean_state = ExternalStatus.DEFAULT193 safe_to_update = stat.safe_to_update()194 self.assertFalse(safe_to_update)195 stat.clean_state = ExternalStatus.EMPTY196 safe_to_update = stat.safe_to_update()197 self.assertFalse(safe_to_update)198 stat.clean_state = ExternalStatus.UNKNOWN199 safe_to_update = stat.safe_to_update()200 self.assertFalse(safe_to_update)201 stat.clean_state = ExternalStatus.STATUS_OK202 safe_to_update = stat.safe_to_update()203 self.assertFalse(safe_to_update)204 stat.clean_state = ExternalStatus.DIRTY205 safe_to_update = stat.safe_to_update()206 self.assertFalse(safe_to_update)207 def test_update_empty_all(self):208 """If the repository in-sync is empty, then it is not safe to209 update, regardless of the clean state.210 """211 stat = ExternalStatus()212 stat.sync_state = ExternalStatus.UNKNOWN213 stat.clean_state = ExternalStatus.DEFAULT214 safe_to_update = stat.safe_to_update()215 self.assertFalse(safe_to_update)216 stat.clean_state = ExternalStatus.EMPTY217 safe_to_update = stat.safe_to_update()218 self.assertFalse(safe_to_update)219 stat.clean_state = ExternalStatus.UNKNOWN220 safe_to_update = stat.safe_to_update()221 self.assertFalse(safe_to_update)222 stat.clean_state = ExternalStatus.STATUS_OK223 safe_to_update = stat.safe_to_update()224 self.assertFalse(safe_to_update)225 stat.clean_state = ExternalStatus.DIRTY226 safe_to_update = stat.safe_to_update()227 self.assertFalse(safe_to_update)228if __name__ == '__main__':...
put.test.ts
Source:put.test.ts
1import { EntityManager } from '@mikro-orm/core'2import Koa from 'koa'3import init from '../../../src/index'4import request from 'supertest'5import User from '../../../src/entities/user'6import { genAccessToken } from '../../../src/lib/auth/buildTokenPair'7import UserFactory from '../../fixtures/UserFactory'8import Game from '../../../src/entities/game'9import GameStatFactory from '../../fixtures/GameStatFactory'10import GameFactory from '../../fixtures/GameFactory'11import GameActivity, { GameActivityType } from '../../../src/entities/game-activity'12import casual from 'casual'13describe('Game stat service - put', () => {14 let app: Koa15 let user: User16 let token: string17 let game: Game18 beforeAll(async () => {19 app = await init()20 user = await new UserFactory().one()21 game = await new GameFactory(user.organisation).one()22 await (<EntityManager>app.context.em).persistAndFlush([user, game])23 token = await genAccessToken(user)24 })25 beforeEach(async () => {26 const activities = await (<EntityManager>app.context.em).getRepository(GameActivity).findAll()27 await (<EntityManager>app.context.em).removeAndFlush(activities)28 })29 afterAll(async () => {30 await (<EntityManager>app.context.em).getConnection().close()31 })32 it('should update the name', async () => {33 const stat = await new GameStatFactory([game]).one()34 await (<EntityManager>app.context.em).persistAndFlush(stat)35 const res = await request(app.callback())36 .put(`/games/${game.id}/game-stats/${stat.id}`)37 .send({ internalName: stat.internalName, name: 'New name', global: stat.global, maxChange: stat.maxChange, minValue: stat.minValue, maxValue: stat.maxValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })38 .auth(token, { type: 'bearer' })39 .expect(200)40 expect(res.body.stat.name).toBe('New name')41 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({42 type: GameActivityType.GAME_STAT_UPDATED,43 extra: {44 statInternalName: res.body.stat.internalName45 }46 })47 expect(activity.extra.display).toStrictEqual({48 'Updated properties': 'name: New name'49 })50 })51 it('should update the global status', async () => {52 const stat = await new GameStatFactory([game]).with(() => ({ global: false })).one()53 await (<EntityManager>app.context.em).persistAndFlush(stat)54 const res = await request(app.callback())55 .put(`/games/${game.id}/game-stats/${stat.id}`)56 .send({ global: true, internalName: stat.internalName, name: stat.name, maxChange: stat.maxChange, minValue: stat.minValue, maxValue: stat.maxValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })57 .auth(token, { type: 'bearer' })58 .expect(200)59 expect(res.body.stat.global).toBe(true)60 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({61 type: GameActivityType.GAME_STAT_UPDATED,62 extra: {63 statInternalName: res.body.stat.internalName64 }65 })66 expect(activity.extra.display).toStrictEqual({67 'Updated properties': 'global: true'68 })69 })70 it('should update the max change', async () => {71 const stat = await new GameStatFactory([game]).one()72 await (<EntityManager>app.context.em).persistAndFlush(stat)73 const res = await request(app.callback())74 .put(`/games/${game.id}/game-stats/${stat.id}`)75 .send({ maxChange: 90, internalName: stat.internalName, name: stat.name, global: stat.global, minValue: stat.minValue, maxValue: stat.maxValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })76 .auth(token, { type: 'bearer' })77 .expect(200)78 expect(res.body.stat.maxChange).toBe(90)79 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({80 type: GameActivityType.GAME_STAT_UPDATED,81 extra: {82 statInternalName: res.body.stat.internalName83 }84 })85 expect(activity.extra.display).toStrictEqual({86 'Updated properties': 'maxChange: 90'87 })88 })89 it('should update the min value', async () => {90 const stat = await new GameStatFactory([game]).one()91 await (<EntityManager>app.context.em).persistAndFlush(stat)92 const res = await request(app.callback())93 .put(`/games/${game.id}/game-stats/${stat.id}`)94 .send({ minValue: -300, internalName: stat.internalName, name: stat.name, global: stat.global, maxChange: stat.maxChange, maxValue: stat.maxValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })95 .auth(token, { type: 'bearer' })96 .expect(200)97 expect(res.body.stat.minValue).toBe(-300)98 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({99 type: GameActivityType.GAME_STAT_UPDATED,100 extra: {101 statInternalName: res.body.stat.internalName102 }103 })104 expect(activity.extra.display).toStrictEqual({105 'Updated properties': 'minValue: -300'106 })107 })108 it('should update the max value', async () => {109 const stat = await new GameStatFactory([game]).one()110 await (<EntityManager>app.context.em).persistAndFlush(stat)111 const newMax = stat.maxValue + 80112 const res = await request(app.callback())113 .put(`/games/${game.id}/game-stats/${stat.id}`)114 .send({ maxValue: newMax, internalName: stat.internalName, name: stat.name, global: stat.global, maxChange: stat.maxChange, minValue: stat.minValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })115 .auth(token, { type: 'bearer' })116 .expect(200)117 expect(res.body.stat.maxValue).toBe(newMax)118 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({119 type: GameActivityType.GAME_STAT_UPDATED,120 extra: {121 statInternalName: res.body.stat.internalName122 }123 })124 expect(activity.extra.display).toStrictEqual({125 'Updated properties': `maxValue: ${newMax}`126 })127 })128 it('should update the default value', async () => {129 const stat = await new GameStatFactory([game]).one()130 await (<EntityManager>app.context.em).persistAndFlush(stat)131 const defaultValue = casual.integer(stat.minValue, stat.maxValue)132 const res = await request(app.callback())133 .put(`/games/${game.id}/game-stats/${stat.id}`)134 .send({ defaultValue, internalName: stat.internalName, name: stat.name, global: stat.global, maxChange: stat.maxChange, minValue: stat.minValue, maxValue: stat.maxValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })135 .auth(token, { type: 'bearer' })136 .expect(200)137 expect(res.body.stat.defaultValue).toBe(defaultValue)138 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({139 type: GameActivityType.GAME_STAT_UPDATED,140 extra: {141 statInternalName: res.body.stat.internalName142 }143 })144 expect(activity.extra.display).toStrictEqual({145 'Updated properties': `defaultValue: ${defaultValue}`146 })147 })148 it('should update the min time between updates', async () => {149 const stat = await new GameStatFactory([game]).one()150 await (<EntityManager>app.context.em).persistAndFlush(stat)151 const res = await request(app.callback())152 .put(`/games/${game.id}/game-stats/${stat.id}`)153 .send({ minTimeBetweenUpdates: 10242, internalName: stat.internalName, name: stat.name, global: stat.global, maxChange: stat.maxChange, minValue: stat.minValue, maxValue: stat.maxValue, defaultValue: stat.defaultValue })154 .auth(token, { type: 'bearer' })155 .expect(200)156 expect(res.body.stat.minTimeBetweenUpdates).toBe(10242)157 const activity = await (<EntityManager>app.context.em).getRepository(GameActivity).findOne({158 type: GameActivityType.GAME_STAT_UPDATED,159 extra: {160 statInternalName: res.body.stat.internalName161 }162 })163 expect(activity.extra.display).toStrictEqual({164 'Updated properties': 'minTimeBetweenUpdates: 10242'165 })166 })167 it('should not update a non-existent stat', async () => {168 const stat = await new GameStatFactory([game]).one()169 const res = await request(app.callback())170 .put(`/games/${game.id}/game-stats/31223`)171 .send({ internalName: stat.internalName, name: stat.name, global: stat.global, maxChange: stat.maxChange, minValue: stat.minValue, maxValue: stat.maxValue, defaultValue: stat.defaultValue, minTimeBetweenUpdates: stat.minTimeBetweenUpdates })172 .auth(token, { type: 'bearer' })173 .expect(404)174 expect(res.body).toStrictEqual({ message: 'Stat not found' })175 })...
caching_file_system.py
Source:caching_file_system.py
1# Copyright (c) 2012 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4import posixpath5import sys6from file_system import FileSystem, StatInfo, FileNotFoundError7from future import Future8from path_util import IsDirectory9from third_party.json_schema_compiler.memoize import memoize10class CachingFileSystem(FileSystem):11 '''FileSystem which implements a caching layer on top of |file_system|. It's12 smart, using Stat() to decided whether to skip Read()ing from |file_system|,13 and only Stat()ing directories never files.14 '''15 def __init__(self, file_system, object_store_creator):16 self._file_system = file_system17 def create_object_store(category, **optargs):18 return object_store_creator.Create(19 CachingFileSystem,20 category='%s/%s' % (file_system.GetIdentity(), category),21 **optargs)22 self._stat_object_store = create_object_store('stat')23 # The read caches can start populated (start_empty=False) because file24 # updates are picked up by the stat, so it doesn't need the force-refresh25 # which starting empty is designed for. Without this optimisation, cron26 # runs are extra slow.27 self._read_object_store = create_object_store('read', start_empty=False)28 def Refresh(self):29 return self._file_system.Refresh()30 def Stat(self, path):31 return self.StatAsync(path).Get()32 def StatAsync(self, path):33 '''Stats the directory given, or if a file is given, stats the file's parent34 directory to get info about the file.35 '''36 # Always stat the parent directory, since it will have the stat of the child37 # anyway, and this gives us an entire directory's stat info at once.38 dir_path, file_path = posixpath.split(path)39 if dir_path and not dir_path.endswith('/'):40 dir_path += '/'41 def make_stat_info(dir_stat):42 '''Converts a dir stat into the correct resulting StatInfo; if the Stat43 was for a file, the StatInfo should just contain that file.44 '''45 if path == dir_path:46 return dir_stat47 # Was a file stat. Extract that file.48 file_version = dir_stat.child_versions.get(file_path)49 if file_version is None:50 raise FileNotFoundError('No stat found for %s in %s (found %s)' %51 (path, dir_path, dir_stat.child_versions))52 return StatInfo(file_version)53 dir_stat = self._stat_object_store.Get(dir_path).Get()54 if dir_stat is not None:55 return Future(value=make_stat_info(dir_stat))56 dir_stat_future = self._MemoizedStatAsyncFromFileSystem(dir_path)57 def resolve():58 dir_stat = dir_stat_future.Get()59 assert dir_stat is not None # should have raised a FileNotFoundError60 # We only ever need to cache the dir stat.61 self._stat_object_store.Set(dir_path, dir_stat)62 return make_stat_info(dir_stat)63 return Future(callback=resolve)64 @memoize65 def _MemoizedStatAsyncFromFileSystem(self, dir_path):66 '''This is a simple wrapper to memoize Futures to directory stats, since67 StatAsync makes heavy use of it. Only cache directories so that the68 memoized cache doesn't blow up.69 '''70 assert IsDirectory(dir_path)71 return self._file_system.StatAsync(dir_path)72 def Read(self, paths, skip_not_found=False):73 '''Reads a list of files. If a file is in memcache and it is not out of74 date, it is returned. Otherwise, the file is retrieved from the file system.75 '''76 cached_read_values = self._read_object_store.GetMulti(paths).Get()77 cached_stat_values = self._stat_object_store.GetMulti(paths).Get()78 # Populate a map of paths to Futures to their stat. They may have already79 # been cached in which case their Future will already have been constructed80 # with a value.81 stat_futures = {}82 def swallow_file_not_found_error(future):83 def resolve():84 try: return future.Get()85 except FileNotFoundError: return Nnone86 return Future(callback=resolve)87 for path in paths:88 stat_value = cached_stat_values.get(path)89 if stat_value is None:90 stat_future = self.StatAsync(path)91 if skip_not_found:92 stat_future = swallow_file_not_found_error(stat_future)93 else:94 stat_future = Future(value=stat_value)95 stat_futures[path] = stat_future96 # Filter only the cached data which is fresh by comparing to the latest97 # stat. The cached read data includes the cached version. Remove it for98 # the result returned to callers.99 fresh_data = dict(100 (path, data) for path, (data, version) in cached_read_values.iteritems()101 if stat_futures[path].Get().version == version)102 if len(fresh_data) == len(paths):103 # Everything was cached and up-to-date.104 return Future(value=fresh_data)105 # Read in the values that were uncached or old.106 read_futures = self._file_system.Read(107 set(paths) - set(fresh_data.iterkeys()),108 skip_not_found=skip_not_found)109 def resolve():110 new_results = read_futures.Get()111 # Update the cache. This is a path -> (data, version) mapping.112 self._read_object_store.SetMulti(113 dict((path, (new_result, stat_futures[path].Get().version))114 for path, new_result in new_results.iteritems()))115 new_results.update(fresh_data)116 return new_results117 return Future(callback=resolve)118 def GetIdentity(self):119 return self._file_system.GetIdentity()120 def __repr__(self):...
peer_record.py
Source:peer_record.py
1import math2import time3from dataclasses import dataclass4from chia.util.ints import uint32, uint645from chia.util.streamable import Streamable, streamable6@streamable7@dataclass(frozen=True)8class PeerRecord(Streamable):9 peer_id: str10 ip_address: str11 port: uint3212 connected: bool13 last_try_timestamp: uint6414 try_count: uint3215 connected_timestamp: uint6416 added_timestamp: uint6417 best_timestamp: uint6418 version: str19 handshake_time: uint6420 tls_version: str21 def update_version(self, version, now):22 if version != "undefined":23 object.__setattr__(self, "version", version)24 object.__setattr__(self, "handshake_time", uint64(now))25class PeerStat:26 weight: float27 count: float28 reliability: float29 def __init__(self, weight, count, reliability):30 self.weight = weight31 self.count = count32 self.reliability = reliability33 def update(self, is_reachable: bool, age: int, tau: int):34 f = math.exp(-age / tau)35 self.reliability = self.reliability * f + (1.0 - f if is_reachable else 0.0)36 self.count = self.count * f + 1.037 self.weight = self.weight * f + 1.0 - f38class PeerReliability:39 peer_id: str40 ignore_till: int41 ban_till: int42 stat_2h: PeerStat43 stat_8h: PeerStat44 stat_1d: PeerStat45 stat_1w: PeerStat46 stat_1m: PeerStat47 tries: int48 successes: int49 def __init__(50 self,51 peer_id: str,52 ignore_till: int = 0,53 ban_till: int = 0,54 stat_2h_weight: float = 0.0,55 stat_2h_count: float = 0.0,56 stat_2h_reliability: float = 0.0,57 stat_8h_weight: float = 0.0,58 stat_8h_count: float = 0.0,59 stat_8h_reliability: float = 0.0,60 stat_1d_weight: float = 0.0,61 stat_1d_count: float = 0.0,62 stat_1d_reliability: float = 0.0,63 stat_1w_weight: float = 0.0,64 stat_1w_count: float = 0.0,65 stat_1w_reliability: float = 0.0,66 stat_1m_weight: float = 0.0,67 stat_1m_count: float = 0.0,68 stat_1m_reliability: float = 0.0,69 tries: int = 0,70 successes: int = 0,71 ):72 self.peer_id = peer_id73 self.ignore_till = ignore_till74 self.ban_till = ban_till75 self.stat_2h = PeerStat(stat_2h_weight, stat_2h_count, stat_2h_reliability)76 self.stat_8h = PeerStat(stat_8h_weight, stat_8h_count, stat_8h_reliability)77 self.stat_1d = PeerStat(stat_1d_weight, stat_1d_count, stat_1d_reliability)78 self.stat_1w = PeerStat(stat_1w_weight, stat_1w_count, stat_1w_reliability)79 self.stat_1m = PeerStat(stat_1m_weight, stat_1m_count, stat_1m_reliability)80 self.tries = tries81 self.successes = successes82 def is_reliable(self) -> bool:83 if self.tries > 0 and self.tries <= 3 and self.successes * 2 >= self.tries:84 return True85 if self.stat_2h.reliability > 0.85 and self.stat_2h.count > 2:86 return True87 if self.stat_8h.reliability > 0.7 and self.stat_8h.count > 4:88 return True89 if self.stat_1d.reliability > 0.55 and self.stat_1d.count > 8:90 return True91 if self.stat_1w.reliability > 0.45 and self.stat_1w.count > 16:92 return True93 if self.stat_1m.reliability > 0.35 and self.stat_1m.count > 32:94 return True95 return False96 def get_ban_time(self) -> int:97 if self.is_reliable():98 return 099 if self.stat_1m.reliability - self.stat_1m.weight + 1 < 0.15 and self.stat_1m.count > 32:100 return 30 * 86400101 if self.stat_1w.reliability - self.stat_1w.weight + 1.0 < 0.10 and self.stat_1w.count > 16:102 return 7 * 86400103 if self.stat_1d.reliability - self.stat_1d.weight + 1.0 < 0.05 and self.stat_1d.count > 8:104 return 86400105 return 0106 def get_ignore_time(self) -> int:107 if self.is_reliable():108 return 0109 if self.stat_1m.reliability - self.stat_1m.weight + 1.0 < 0.20 and self.stat_1m.count > 2:110 return 10 * 86400111 if self.stat_1w.reliability - self.stat_1w.weight + 1.0 < 0.16 and self.stat_1w.count > 2:112 return 3 * 86400113 if self.stat_1d.reliability - self.stat_1d.weight + 1.0 < 0.12 and self.stat_1d.count > 2:114 return 8 * 3600115 if self.stat_8h.reliability - self.stat_8h.weight + 1.0 < 0.08 and self.stat_8h.count > 2:116 return 2 * 3600117 return 0118 def update(self, is_reachable: bool, age: int):119 self.stat_2h.update(is_reachable, age, 2 * 3600)120 self.stat_8h.update(is_reachable, age, 8 * 3600)121 self.stat_1d.update(is_reachable, age, 24 * 3600)122 self.stat_1w.update(is_reachable, age, 7 * 24 * 3600)123 self.stat_1m.update(is_reachable, age, 24 * 30 * 3600)124 self.tries += 1125 if is_reachable:126 self.successes += 1127 current_ignore_time = self.get_ignore_time()128 now = int(time.time())129 if current_ignore_time > 0 and (self.ignore_till == 0 or self.ignore_till < current_ignore_time + now):130 self.ignore_till = current_ignore_time + now131 current_ban_time = self.get_ban_time()132 if current_ban_time > 0 and (self.ban_till == 0 or self.ban_till < current_ban_time + now):...
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 lisa 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?
