Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use mk_cgroup method in autotest

Best Python code snippet using autotest_python

cgroup_common.py

Source:cgroup_common.py

...34        else:35            logging.error("cg.initialize(): Module %s not found", self.module)36            return -137        return 038    def mk_cgroup(self, root=None):39        """40        Creates new temporary cgroup41        @param root: where to create this cgroup (default: self.root)42        @return: 0 when PASSED43        """44        try:45            if root:46                pwd = mkdtemp(prefix='cgroup-', dir=root) + '/'47            else:48                pwd = mkdtemp(prefix='cgroup-', dir=self.root) + '/'49        except Exception, inst:50            logging.error("cg.mk_cgroup(): %s" , inst)51            return None52        return pwd53    def rm_cgroup(self, pwd, supress=False):54        """55        Removes cgroup.56        @param pwd: cgroup directory.57        @param supress: supress output.58        @return: 0 when PASSED59        """60        try:61            os.rmdir(pwd)62        except Exception, inst:63            if not supress:64                logging.error("cg.rm_cgroup(): %s" , inst)65            return -166        return 067    def test(self, cmd):68        """69        Executes cgroup_client.py with cmd parameter.70        @param cmd: command to be executed71        @return: subprocess.Popen() process72        """73        logging.debug("cg.test(): executing paralel process '%s'", cmd)74        cmd = self._client + ' ' + cmd75        process = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE,76                                   stdout=subprocess.PIPE,77                                   stderr=subprocess.PIPE, close_fds=True)78        return process79    def is_cgroup(self, pid, pwd):80        """81        Checks if the 'pid' process is in 'pwd' cgroup82        @param pid: pid of the process83        @param pwd: cgroup directory84        @return: 0 when is 'pwd' member85        """86        if open(pwd + '/tasks').readlines().count("%d\n" % pid) > 0:87            return 088        else:89            return -190    def is_root_cgroup(self, pid):91        """92        Checks if the 'pid' process is in root cgroup (WO cgroup)93        @param pid: pid of the process94        @return: 0 when is 'root' member95        """96        return self.is_cgroup(pid, self.root)97    def set_cgroup(self, pid, pwd):98        """99        Sets cgroup membership100        @param pid: pid of the process101        @param pwd: cgroup directory102        @return: 0 when PASSED103        """104        try:105            open(pwd+'/tasks', 'w').write(str(pid))106        except Exception, inst:107            logging.error("cg.set_cgroup(): %s" , inst)108            return -1109        if self.is_cgroup(pid, pwd):110            logging.error("cg.set_cgroup(): Setting %d pid into %s cgroup "111                          "failed", pid, pwd)112            return -1113        else:114            return 0115    def set_root_cgroup(self, pid):116        """117        Resets the cgroup membership (sets to root)118        @param pid: pid of the process119        @return: 0 when PASSED120        """121        return self.set_cgroup(pid, self.root)122    def get_prop(self, prop, pwd=None, supress=False):123        """124        Gets one line of the property value125        @param prop: property name (file)126        @param pwd: cgroup directory127        @param supress: supress the output128        @return: String value or None when FAILED129        """130        tmp = self.get_property(prop, pwd, supress)131        if tmp:132            if tmp[0][-1] == '\n':133                tmp[0] = tmp[0][:-1]134            return tmp[0]135        else:136            return None137    def get_property(self, prop, pwd=None, supress=False):138        """139        Gets the property value140        @param prop: property name (file)141        @param pwd: cgroup directory142        @param supress: supress the output143        @return: [] values or None when FAILED144        """145        if pwd == None:146            pwd = self.root147        try:148            ret = open(pwd+prop, 'r').readlines()149        except Exception, inst:150            ret = None151            if not supress:152                logging.error("cg.get_property(): %s" , inst)153        return ret154    def set_prop(self, prop, value, pwd=None, check=True):155        """156        Sets the one-line property value concerning the K,M,G postfix157        @param prop: property name (file)158        @param value: desired value159        @param pwd: cgroup directory160        @param check: check the value after setup161        @return: 0 when PASSED162        """163        _value = value164        try:165            value = str(value)166            if value[-1] == '\n':167                value = value[:-1]168            if value[-1] == 'K':169                value = int(value[:-1]) * 1024170            elif value[-1] == 'M':171                value = int(value[:-1]) * 1048576172            elif value[-1] == 'G':173                value = int(value[:-1]) * 1073741824174        except:175            logging.error("cg.set_prop() fallback into cg.set_property.")176            value = _value177        return self.set_property(prop, value, pwd, check)178    def set_property(self, prop, value, pwd=None, check=True):179        """180        Sets the property value181        @param prop: property name (file)182        @param value: desired value183        @param pwd: cgroup directory184        @param check: check the value after setup185        @return: 0 when PASSED186        """187        value = str(value)188        if pwd == None:189            pwd = self.root190        try:191            open(pwd+prop, 'w').write(value)192        except Exception, inst:193            logging.error("cg.set_property(): %s" , inst)194            return -1195        if check:196            # Get the first line - '\n'197            _value = self.get_property(prop, pwd)[0][:-1]198            if value != _value:199                logging.error("cg.set_property(): Setting failed: desired = %s,"200                              " real value = %s", value, _value)201                return -1202        return 0203    def smoke_test(self):204        """205        Smoke test206        Module independent basic tests207        """208        part = 0209        pwd = self.mk_cgroup()210        if pwd == None:211            logging.error("cg.smoke_test[%d]: Can't create cgroup", part)212            return -1213        part += 1214        ps = self.test("smoke")215        if ps == None:216            logging.error("cg.smoke_test[%d]: Couldn't create process", part)217            return -1218        part += 1219        if (ps.poll() != None):220            logging.error("cg.smoke_test[%d]: Process died unexpectidly", part)221            return -1222        # New process should be a root member223        part += 1...

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.