m2-traitement-benchmark/interfaces/db_testing_interface.py

from typing import Generic, TypeVar
from typing_extensions import Dict
from abc import abstractmethod
from utils.utils import time_func
from time import sleep

T = TypeVar('T')
NUM_RUNS = 100

class DB_Testing(Generic[T]):
    driver_name: str
    
    @abstractmethod
    def __init__(self) -> None:
        pass
    
    @abstractmethod
    def delete_table(self, driver: T):
        pass
    
    @abstractmethod
    def create_table(self, driver: T):
        pass
    
    @abstractmethod
    def delete_data(self, driver: T):
        pass
    
    @abstractmethod
    def add_bulk(self, driver: T):
        pass
    
    @abstractmethod
    def add_singles(self, driver: T):
        pass

    @abstractmethod
    def attach5_mr_mrs(self, driver: T) -> str:
        pass
    
    @abstractmethod
    def search_mails_ends_microsoftcom(self, driver: T) -> str:
        pass
    
    @abstractmethod
    def search_lorem(self, driver: T) -> str:
        pass
    
    @abstractmethod
    def get_sum_attachment_less_5(self, driver: T) -> str:
        pass
    
    @abstractmethod
    def update_add_replies_per_last_name(self, driver: T):
        pass
    
    @abstractmethod
    def update_add_replies_per_subject(self, driver: T):
        pass
    
    def do_tests(self, driver: T) -> Dict[str, float]:
        stats = {}
        print(f"\n### Starting {self.driver_name} Benchmark ###")
        self.delete_table(driver)
        self.create_table(driver)
        stats.update(time_func(f"[{self.driver_name}] Add singles", "add_singles", lambda : self.add_singles(driver)))
        stats.update(time_func(f"[{self.driver_name}] Delete data", "delete_data", lambda : self.delete_data(driver)))
        stats.update(time_func(f"[{self.driver_name}] Add bulk", "add_bulk", lambda : self.add_bulk(driver)))
        
        sleep(1) # To wait for any indexing to avoid race conditions somehow
        
        # Multiple
        
        # Search
        stats.update(time_func(f"[{self.driver_name}] Count mails with more than 5 attachments and Mr/Mrs", "attach5_mr_mrs", lambda : self.attach5_mr_mrs(driver), NUM_RUNS))
        stats.update(time_func(f"[{self.driver_name}] Sum the number of attachment from mail that have less than 5 attachments", "get_sum_attachment_less_5", lambda : self.get_sum_attachment_less_5(driver), NUM_RUNS))
        stats.update(time_func(f"[{self.driver_name}] Count mails that have a lorem sentence", "search_lorem", lambda : self.search_lorem(driver), NUM_RUNS))
        stats.update(time_func(f"[{self.driver_name}] Count mails that have a mail that ends with microsoft.com", "search_mails_ends_microsoftcom", lambda : self.search_mails_ends_microsoftcom(driver), NUM_RUNS))
        
        # Update
        stats.update(time_func(f"[{self.driver_name}] Update all thread to add a field with the total replies per last name", "update_add_replies_per_last_name", lambda : self.update_add_replies_per_last_name(driver), NUM_RUNS))
        stats.update(time_func(f"[{self.driver_name}] Update all thread to add a field with the total replies per subject", "update_add_replies_per_subject", lambda : self.update_add_replies_per_subject(driver), NUM_RUNS))


        print("\n")
        return stats