KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_uuid method in yandex-tank
Best Python code snippet using yandex-tank
models.py
Source:models.py 
...25 self.set_password(password)26 now = datetime.utcnow()27 self.created_at = now28 self.updated_at = now29 self.set_uuid()30 def __repr__(self):31 return '<User {} - {} ({})>'.format(self.username, self.email, self.uuid)32 def set_uuid(self):33 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.email))34 def set_password(self, password):35 self.password_hash = generate_password_hash(password)36 def check_password(self, password):37 return check_password_hash(self.password_hash, password)38 def to_dict(self):39 data = {40 "uuid": self.uuid,41 "username": self.username,42 "email": self.email,43 "_meta": {44 "self": "URL FOR THE USER"45 }46 }47 return data48 def create_from_dict(self, data):49 for field in ['username', 'email']:50 setattr(self, field, data[field])51 timestamp = datetime.utcnow()52 for field in ['created_at', 'updated_at']:53 setattr(self, field, timestamp)54 self.set_uuid()55 if 'password' in data:56 self.set_password(data['password'])57class Transaction(db.Model):58 id = db.Column(db.Integer, primary_key=True)59 uuid = db.Column(db.String(36), index=True, unique=True)60 created_at = db.Column(db.DateTime, index=True, nullable=False)61 updated_at = db.Column(db.DateTime, index=True, nullable=False)62 timestamp = db.Column(db.DateTime, index=True, nullable=False)63 status_id = db.Column(db.ForeignKey('trade_status.id'))64 user_id = db.Column(db.ForeignKey('user.id'))65 position_id = db.Column(db.ForeignKey('trade_position.id'))66 strategy_id = db.Column(db.ForeignKey('trade_strategy.id'))67 account_id = db.Column(db.ForeignKey('trade_account.id'))68 security_id = db.Column(db.ForeignKey('security.id'))69 trade_id = db.Column(db.ForeignKey('trade.id'))70 notes = db.relationship('TradeNote', backref=backref("transaction", lazy="joined"))71 @property72 def sign(self):73 return self._sign74class Trade(db.Model):75 """A Trade consists of one or more transactions"""76 id = db.Column(db.Integer, primary_key=True)77 uuid = db.Column(db.String(36), index=True, unique=True)78 created_at = db.Column(db.DateTime, index=True, nullable=False)79 updated_at = db.Column(db.DateTime, index=True, nullable=False)80 entry_timestamp = db.Column(db.DateTime, index=True, nullable=False)81 entry_price = db.Column(db.Float, nullable=False)82 exit_timestamp = db.Column(db.DateTime, index=True)83 exit_price = db.Column(db.Float)84 shares = db.Column(db.Integer(), nullable=False)85 status_id = db.Column(db.ForeignKey('trade_status.id'))86 user_id = db.Column(db.ForeignKey('user.id'))87 position_id = db.Column(db.ForeignKey('trade_position.id'))88 strategy_id = db.Column(db.ForeignKey('trade_strategy.id'))89 account_id = db.Column(db.ForeignKey('trade_account.id'))90 security_id = db.Column(db.ForeignKey('security.id'))91 notes = db.relationship('TradeNote', backref=backref("trade", lazy="joined"))92 transactions = db.relationship('Transaction', backref=backref("trade", lazy="joined"))93 # def __init__(self, user, account, security, position, shares, entry_time, entry_price, strategy):94 # # TODO: learn how to create multiple things at once in SQLAlchemy95 #96 # self.user = user97 # self.account = account98 # self.security = security99 # self.position = position100 # self.shares = int(shares)101 # self.entry_timestamp = entry_time102 # self.entry_price = float(entry_price)103 # self.strategy = strategy104 #105 # now = datetime.utcnow()106 # self.created_at = now107 # self.updated_at = now108 # self.set_uuid()109 def __repr__(self):110 # <Trade [+2 AYX @ 180.55] (85f94948-c1e0-5df1-bb65-068a31e1bcd4)>111 return '<Trade [{}{} {} @ {} ({})]>'.format(self.sign, self.shares, self.security.ticker, self.entry_price, self.uuid)112 def set_uuid(self):113 uuid_parents = [114 self.user.uuid,115 self.account.broker.uuid,116 self.account.uuid,117 self.security.uuid,118 self.entry_timestamp,119 self.entry_price,120 self.shares121 ]122 self.uuid = str(self.build_uuid(uuid_parents))123 def build_uuid(self, uuids):124 # This loop should build the final uuid based on the parent UUIDs in a cannonical way125 # TODO: make sure this works correctly126 temp_uuid = None127 for i, item in enumerate(uuids):128 if i == 0:129 temp_uuid = uuid.uuid5(NAMESPACE_GPFIX, str(item))130 else:131 temp_uuid = uuid.uuid5(temp_uuid, str(item))132 return temp_uuid133 def to_dict(self):134 data = {135 "uuid": self.uuid,136 "user": self.user.to_dict(),137 "_meta": {138 "self": "URL FOR THE TRADE"139 }140 }141 return data142 def create_from_dict(self, data):143 # ensure required fields are in the data144 required_fields = ['user', 'account', 'ticker', 'shares', 'entry_timestamp', 'entry_price']145 for field in required_fields:146 if field not in data:147 raise ValueError('{} is required'.format(field))148 if not data[field]:149 raise ValueError('{} cannot be empty!'.format(field))150 with db.session.no_autoflush:151 # TODO: handle UUID152 user = User.query.filter_by(username=data['user']).first()153 if user is None:154 raise ValueError('user does not exist')155 # TODO: Validate request for user156 # elif user is not current_user:157 # raise ValueError('403 Forbidden')158 else:159 setattr(self, 'user', user)160 # TODO: handle UUID161 account = TradeAccount.query.filter_by(name=data['account']).first()162 if account is None:163 raise ValueError('account does not exist')164 else:165 setattr(self, 'account', account)166 security = Security.query.filter_by(ticker=data['ticker']).first()167 if security is None:168 # Add the new security169 security = Security()170 security.create_from_symbol(data['ticker'])171 setattr(self, 'security', security)172 # TODO: check for int type?173 setattr(self, 'shares', int(data['shares']))174 if self.shares > 0:175 self.position = TradePosition.query.filter_by(position="LONG").one()176 elif self.shares < 0:177 self.position = TradePosition.query.filter_by(position="SHORT").one()178 else:179 raise ValueError("Invalid entry for shares: {}".format(self.shares))180 if isinstance(data['entry_timestamp'].date(), date):181 setattr(self, 'entry_timestamp', data['entry_timestamp'])182 else:183 setattr(self, 'entry_timestamp', datetime.strptime(data['entry_timestamp'], TIME_FORMAT))184 setattr(self, 'entry_price', float(data['entry_price']))185 setattr(self, 'status', TradeStatus.query.filter_by(status='OPEN').one())186 # Optionals187 if 'exit_timestamp' in data:188 if 'exit_price' not in data:189 raise ValueError("When exit_timestamp is present, exit_price is required")190 setattr(self, 'exit_timestamp', datetime.strptime(data['exit_timestamp'], TIME_FORMAT))191 setattr(self, 'status', TradeStatus.query.filter_by('CLOSED').one())192 if 'exit_price' in data:193 if 'exit_timestamp' not in data:194 raise ValueError("When exit_price is present, exit_timestamp is required")195 setattr(self, 'exit_price', float(data['exit_price']))196 if 'strategy' in data:197 # TODO: filter_by(user_id=current_user.id)198 strategy = TradeStrategy.query.filter_by(user=self.user).filter_by(name=data["strategy"]).one()199 if strategy is None:200 raise ValueError("Strategy [{}] does not exist".format(data['strategy']))201 setattr(self, 'strategy', strategy)202 timestamp = datetime.utcnow()203 for field in ['created_at', 'updated_at']:204 setattr(self, field, timestamp)205 self.set_uuid()206 def close_trade(self, trade_uuid, exit_timestamp, exit_price):207 trade = Trade.query.filter_by(uuid=trade_uuid).first()208 if trade is None:209 raise ValueError("Trade not found. UUID:{}".format(trade_uuid))210 if trade.status.status == "CLOSED":211 raise ValueError("Trade already closed. UUID:{}".format(trade_uuid))212 if trade.status.status != "OPEN":213 raise ValueError("Trade status is {}. {}".format(trade.status.status, trade))214 # zero out the shares215 trade.shares = trade.shares - trade.shares216 trade.exit_timestamp = exit_timestamp217 trade.exit_price = exit_price218 # change the trade status to "CLOSED"219 closed = TradeStatus.query.filter_by(status="CLOSED").first()220 setattr(trade, "status", closed.status)221 trade.updated_at = datetime.utcnow()222 # commit our changes223 db.session.commit()224class Security(db.Model):225 id = db.Column(db.Integer, primary_key=True)226 uuid = db.Column(db.String(36), index=True, unique=True)227 ticker = db.Column(db.String(4), index=True, unique=True)228 full_name = db.Column(db.String(128), index=True, unique=True)229 exchange_id = db.Column(db.ForeignKey('exchange.id'))230 trades = db.relationship('Trade', backref=backref("security", lazy="joined"))231 # def __init__(self, ticker, full_name, exchange):232 # self.ticker = ticker233 # self.full_name = full_name234 # self.exchange = exchange235 # self.set_uuid()236 def __repr__(self):237 return '<Security {} - {} ({})>'.format(self.ticker, self.full_name, self.uuid)238 def set_uuid(self):239 # TODO generate based on exchange UUID240 self.uuid = str(uuid.uuid5(uuid.UUID(self.exchange.uuid), self.ticker))241 def create_from_symbol(self, ticker):242 # https://query2.finance.yahoo.com/v7/finance/options/AAPL243 r = requests.get("https://query2.finance.yahoo.com/v7/finance/options/{}".format(ticker))244 if r.ok:245 data = r.json()246 else:247 raise ValueError("Ticker [{}] not found".format(ticker))248 self.ticker = data['optionChain']['result'][0]['quote']['symbol']249 self.full_name = data['optionChain']['result'][0]['quote']['longName']250 exchange_name = data['optionChain']['result'][0]['quote']['fullExchangeName']251 if re.match(r'^Nasdaq.+$', exchange_name):252 exchange_name = "NASDAQ"253 exchange = Exchange.query.filter_by(name=exchange_name).first()254 if exchange is None:255 exchange = Exchange(name=exchange_name)256 setattr(self, 'exchange', exchange)257 self.set_uuid()258class Exchange(db.Model):259 id = db.Column(db.Integer, primary_key=True)260 uuid = db.Column(db.String(36), index=True, unique=True)261 name = db.Column(db.String(16), index=True, unique=True)262 securities = db.relationship('Security', backref=backref("exchange", lazy="joined"))263 def __init__(self, name):264 self.name = name265 self.set_uuid()266 def __repr__(self):267 return '<Exchange {} ({})>'.format(self.name, self.uuid)268 def set_uuid(self):269 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.name))270class TradeStatus(db.Model):271 # OPEN, CLOSED272 id = db.Column(db.Integer, primary_key=True)273 uuid = db.Column(db.String(36), index=True, unique=True)274 status = db.Column(db.String(32), index=True, unique=True)275 description = db.Column(db.String(256))276 trades = db.relationship('Trade', backref=backref("status", lazy="joined"))277 def __init__(self, status, description=None):278 self.status = status279 self.description = description280 self.set_uuid()281 def __repr__(self):282 return '<TradeStatus {} ({})>'.format(self.status, self.uuid)283 def set_uuid(self):284 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.status))285class TradeAccount(db.Model):286 id = db.Column(db.Integer, primary_key=True)287 uuid = db.Column(db.String(36), index=True, unique=True)288 created_at = db.Column(db.DateTime, index=True, nullable=False)289 updated_at = db.Column(db.DateTime, index=True, nullable=False)290 name = db.Column(db.String(32), index=True, unique=True)291 account_type_id = db.Column(db.ForeignKey('trade_account_type.id'))292 broker_id = db.Column(db.ForeignKey('broker.id'))293 user_id = db.Column(db.ForeignKey('user.id'))294 trades = db.relationship('Trade', backref=backref("account", lazy="joined"))295 def __init__(self, name, account_type, broker, user):296 self.name = name297 self.type = account_type,298 self.broker = broker299 self.user = user300 now = datetime.utcnow()301 self.created_at = now302 self.updated_at = now303 self.set_uuid()304 def __repr__(self):305 return '<TradeAccount {} ({})>'.format(self.name, self.uuid)306 def set_uuid(self):307 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.name))308class TradeAccountType(db.Model):309 # Cash / Margin310 id = db.Column(db.Integer, primary_key=True)311 uuid = db.Column(db.String(36), index=True, unique=True)312 type = db.Column(db.String(32), index=True, unique=True)313 accounts = db.relationship('TradeAccount', backref=backref("account", lazy="joined"))314 def __init__(self, type):315 self.type = type316 self.set_uuid()317 def __repr__(self):318 return '<TradeAccountType {} ({})>'.format(self.type, self.uuid)319 def set_uuid(self):320 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.type))321class Broker(db.Model):322 id = db.Column(db.Integer, primary_key=True)323 uuid = db.Column(db.String(36), index=True, unique=True)324 created_at = db.Column(db.DateTime, index=True, nullable=False)325 updated_at = db.Column(db.DateTime, index=True, nullable=False)326 name = db.Column(db.String(32), index=True, unique=True)327 accounts = db.relationship('TradeAccount', backref=backref("broker", lazy="joined"))328 def __init__(self, name):329 self.name = name330 now = datetime.utcnow()331 self.created_at = now332 self.updated_at = now333 self.set_uuid()334 def __repr__(self):335 return '<Broker {} ({})>'.format(self.name, self.uuid)336 def set_uuid(self):337 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.name))338class TradePosition(db.Model):339 # LONG / SHORT340 id = db.Column(db.Integer, primary_key=True)341 uuid = db.Column(db.String(36), index=True, unique=True)342 position = db.Column(db.String(32), index=True, unique=True)343 trades = db.relationship('Trade', backref=backref("position", lazy="joined"))344 def __init__(self, position):345 self.position = position346 self.set_uuid()347 def __repr__(self):348 return '<TradePosition {} ({})>'.format(self.position, self.uuid)349 def set_uuid(self):350 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.position))351class TradeStrategy(db.Model):352 id = db.Column(db.Integer, primary_key=True)353 uuid = db.Column(db.String(36), index=True, unique=True)354 created_at = db.Column(db.DateTime, index=True, nullable=False)355 updated_at = db.Column(db.DateTime, index=True, nullable=False)356 name = db.Column(db.String(64), index=True, unique=True)357 description = db.Column(db.String(256), index=True, unique=True)358 user_id = db.Column(db.ForeignKey('user.id'))359 trades = db.relationship('Trade', backref=backref("strategy", lazy="joined"))360 def __init__(self, name, description, user):361 self.name = name362 self.description = description363 self.user = user364 now = datetime.utcnow()365 self.created_at = now366 self.updated_at = now367 self.set_uuid()368 def __repr__(self):369 return '<TradeStrategy {} ({})>'.format(self.name, self.uuid)370 def set_uuid(self):371 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.name))372class TradeNote(db.Model):373 id = db.Column(db.Integer, primary_key=True)374 uuid = db.Column(db.String(36), index=True, unique=True)375 created_at = db.Column(db.DateTime, index=True, nullable=False)376 updated_at = db.Column(db.DateTime, index=True, nullable=False)377 note = db.Column(db.String(256))378 trade_id = db.Column(db.ForeignKey('trade.id'))379 def __init__(self, note):380 self.note = note381 now = datetime.utcnow()382 self.created_at = now383 self.updated_at = now384 self.set_uuid()385 def __repr__(self):386 return '<TradeNote {}>'.format(self.uuid)387 def set_uuid(self):388 self.uuid = str(uuid.uuid5(NAMESPACE_GPFIX, self.note))389@login.user_loader390def load_user(id):...
solr_index.py
Source:solr_index.py
1import collections2import pickle3import time4from typing import Deque, Hashable5from smqtk.representation.descriptor_set import DescriptorSet6# Try to import required module7try:8 import solr # type: ignore9except ImportError:10 solr = None11__author__ = "paul.tunison@kitware.com"12class SolrDescriptorSet (DescriptorSet):13 """14 Descriptor set that uses a Solr instance as a backend storage medium.15 Fields where components are stored within a document are specified at16 construction time. We optionally set the ``id`` field to a string UUID.17 ``id`` is set because it is a common, required field for unique18 identification of documents.19 Descriptor UUIDs should maintain their uniqueness when converted to a20 string, otherwise this backend will not work well when querying.21 """22 @classmethod23 def is_usable(cls):24 return solr is not None25 def __init__(self, solr_conn_addr, set_uuid,26 set_uuid_field, d_uid_field, descriptor_field,27 timestamp_field, solr_params=None,28 commit_on_add=True, max_boolean_clauses=1024,29 pickle_protocol=-1):30 """31 Construct a descriptor set pointing to a Solr instance.32 :param solr_conn_addr: HTTP(S) address for the Solr set to use33 :type solr_conn_addr: str34 :param set_uuid: Unique ID for the descriptor set to use within the35 configured Solr set.36 :type set_uuid: str37 :param set_uuid_field: Solr set field to store/locate set UUID38 value.39 :type set_uuid_field: str40 :param d_uid_field: Solr set field to store/locate descriptor UUID41 values42 :type d_uid_field: str43 :param descriptor_field: Solr set field to store the code-associated44 descriptor object.45 :type descriptor_field: str46 :param timestamp_field: Solr set field to store floating-point UNIX47 timestamps.48 :type timestamp_field: str49 :param solr_params: Dictionary of additional keyword parameters to set50 in the ``solr.Solr`` instance used. See the ``pysolr``51 documentation for available parameters and values.52 :type solr_params: dict[str, object]53 :param commit_on_add: Immediately commit changes when one or many54 descriptor are added.55 :type commit_on_add: bool56 :param max_boolean_clauses: Solr instance's configured57 maxBooleanClauses configuration property (found in solrconfig.xml58 file). This is needed so we can correctly chunk up batch queries59 without breaking the server. This may also be less than the Solr60 instance's set value.61 :type max_boolean_clauses: int62 :param pickle_protocol: Pickling protocol to use. We will use -1 by63 default (latest version, probably binary).64 :type pickle_protocol: int65 """66 super(SolrDescriptorSet, self).__init__()67 self.set_uuid = set_uuid68 self.set_uuid_field = set_uuid_field69 self.d_uid_field = d_uid_field70 self.descriptor_field = descriptor_field71 self.timestamp_field = timestamp_field72 self.commit_on_add = commit_on_add73 self.max_boolean_clauses = int(max_boolean_clauses)74 assert self.max_boolean_clauses >= 2, "Need more clauses"75 self.pickle_protocol = pickle_protocol76 self.solr_params = solr_params77 self.solr = solr.Solr(solr_conn_addr, **solr_params)78 def __getstate__(self):79 return self.get_config()80 def __setstate__(self, state):81 state['solr'] = solr.Solr(state["solr_conn_addr"],82 **state['solr_params'])83 del state['solr_conn_addr']84 self.__dict__.update(state)85 def _doc_for_code_descr(self, d):86 """87 Generate standard identifying document base for the given88 descriptor element.89 """90 uuid = d.uuid()91 return {92 'id': '-'.join([self.set_uuid, uuid]),93 self.set_uuid_field: self.set_uuid,94 self.d_uid_field: uuid,95 }96 def get_config(self):97 return {98 "solr_conn_addr": self.solr.url,99 "set_uuid": self.set_uuid,100 "set_uuid_field": self.set_uuid_field,101 "d_uid_field": self.d_uid_field,102 "descriptor_field": self.descriptor_field,103 "timestamp_field": self.timestamp_field,104 "solr_params": self.solr_params,105 "commit_on_add": self.commit_on_add,106 "max_boolean_clauses": self.max_boolean_clauses,107 "pickle_protocol": self.pickle_protocol,108 }109 def count(self):110 """111 :return: Number of descriptor elements stored in this set.112 :rtype: int113 """114 return int(self.solr.115 select("%s:%s AND %s:*"116 % (self.set_uuid_field, self.set_uuid,117 self.descriptor_field))118 .numFound)119 def clear(self):120 """121 Clear this descriptor set's entries.122 """123 self.solr.delete_query("%s:%s"124 % (self.set_uuid_field, self.set_uuid))125 self.solr.commit()126 def has_descriptor(self, uuid):127 """128 Check if a DescriptorElement with the given UUID exists in this set.129 :param uuid: UUID to query for130 :type uuid: collections.abc.Hashable131 :return: True if a DescriptorElement with the given UUID exists in this132 set, or False if not.133 :rtype: bool134 """135 # Try to select the descriptor136 # TODO: Probably a better way of doing this that's more efficient.137 return bool(138 self.solr.select("%s:%s AND %s:%s"139 % (self.set_uuid_field, self.set_uuid,140 self.d_uid_field, uuid)).numFound141 )142 def add_descriptor(self, descriptor):143 """144 Add a descriptor to this set.145 Adding the same descriptor multiple times should not add multiple copies146 of the descriptor in the set (based on UUID). Added descriptors147 overwrite set descriptors based on UUID.148 :param descriptor: Descriptor to add to this set.149 :type descriptor: smqtk.representation.DescriptorElement150 """151 doc = self._doc_for_code_descr(descriptor)152 doc[self.descriptor_field] = pickle.dumps(descriptor,153 self.pickle_protocol)154 doc[self.timestamp_field] = time.time()155 self.solr.add(doc, commit=self.commit_on_add)156 def add_many_descriptors(self, descriptors):157 """158 Add multiple descriptors at one time.159 Adding the same descriptor multiple times should not add multiple copies160 of the descriptor in the set (based on UUID). Added descriptors161 overwrite set descriptors based on UUID.162 :param descriptors: Iterable of descriptor instances to add to this163 set.164 :type descriptors:165 collections.abc.Iterable[smqtk.representation.DescriptorElement]166 """167 documents = []168 for d in descriptors:169 doc = self._doc_for_code_descr(d)170 doc[self.descriptor_field] = pickle.dumps(d, self.pickle_protocol)171 doc[self.timestamp_field] = time.time()172 documents.append(doc)173 self.solr.add_many(documents)174 if self.commit_on_add:175 self.solr.commit()176 def get_descriptor(self, uuid):177 """178 Get the descriptor in this set that is associated with the given UUID.179 :param uuid: UUID of the DescriptorElement to get.180 :type uuid: collections.abc.Hashable181 :raises KeyError: The given UUID doesn't associate to a182 DescriptorElement in this set.183 :return: DescriptorElement associated with the queried UUID.184 :rtype: smqtk.representation.DescriptorElement185 """186 return tuple(self.get_many_descriptors(uuid))[0]187 def get_many_descriptors(self, *uuids):188 """189 Get an iterator over descriptors associated to given descriptor UUIDs.190 :param uuids: Iterable of descriptor UUIDs to query for.191 :type uuids: collections.abc.Hashable192 :raises KeyError: A given UUID doesn't associate with a193 DescriptorElement in this set.194 :return: Iterator of descriptors associated 1-to-1 to given uuid values.195 :rtype: collections.abc.Iterable[smqtk.representation.DescriptorElement]196 """197 # Chunk up query based on max clauses available to us198 def batch_query(_batch):199 """200 :param _batch: Batch of UIDs to select.201 :type _batch: list[collections.abc.Hashable]202 """203 query = ' OR '.join([self.d_uid_field + (':%s' % _uid)204 for _uid in _batch])205 r = self.solr.select("%s:%s AND (%s)"206 % (self.set_uuid_field, self.set_uuid,207 query))208 # result batches come in chunks of 10209 for doc in r.results:210 yield pickle.loads(doc[self.descriptor_field])211 for j in range(r.numFound // 10):212 r = r.next_batch()213 for doc in r.results:214 yield pickle.loads(doc[self.descriptor_field])215 batch = []216 for uid in uuids:217 batch.append(uid)218 # Will end up using max_clauses-1 OR statements, and one AND219 if len(batch) == self.max_boolean_clauses:220 for d in batch_query(batch):221 yield d222 batch = []223 # tail batch224 if batch:225 assert len(batch) < self.max_boolean_clauses226 for d in batch_query(batch):227 yield d228 def remove_descriptor(self, uuid):229 """230 Remove a descriptor from this set by the given UUID.231 :param uuid: UUID of the DescriptorElement to remove.232 :type uuid: collections.abc.Hashable233 :raises KeyError: The given UUID doesn't associate to a234 DescriptorElement in this set.235 """236 self.remove_many_descriptors([uuid])237 def remove_many_descriptors(self, uuids):238 """239 Remove descriptors associated to given descriptor UUIDs from this set.240 :param uuids: Iterable of descriptor UUIDs to remove.241 :type uuids: collections.abc.Iterable[collections.abc.Hashable]242 :raises KeyError: A given UUID doesn't associate with a243 DescriptorElement in this set.244 """245 # Chunk up operation based on max clauses available to us246 def batch_op(_batch):247 """248 :param _batch: UIDs to remove from set.249 :type _batch: collections.abc.Iterable[collections.abc.Hashable]250 """251 uuid_query = ' OR '.join([self.d_uid_field + (':%s' % str(_uid))252 for _uid in _batch])253 self.solr.delete("%s:%s AND (%s)"254 % (self.set_uuid_field, self.set_uuid,255 uuid_query))256 batch: Deque[Hashable] = collections.deque()257 for uid in uuids:258 batch.append(uid)259 # Will end up using max_clauses-1 OR statements, and one AND260 if len(batch) == self.max_boolean_clauses:261 batch_op(batch)262 batch.clear()263 # tail batch264 if batch:265 batch_op(batch)266 def iterkeys(self):267 """268 Return an iterator over set descriptor keys, which are their UUIDs.269 """270 r = self.solr.select('%s:%s %s:*'271 % (self.set_uuid_field, self.set_uuid,272 self.d_uid_field))273 for doc in r.results:274 yield doc[self.d_uid_field]275 for _ in range(r.numFound // 10):276 r = r.next_batch()277 for doc in r.results:278 yield doc[self.d_uid_field]279 def iterdescriptors(self):280 """281 Return an iterator over set descriptor element instances.282 """283 r = self.solr.select('%s:%s %s:*'284 % (self.set_uuid_field, self.set_uuid,285 self.descriptor_field))286 for doc in r.results:287 yield pickle.loads(doc[self.descriptor_field])288 for _ in range(r.numFound // 10):289 r = r.next_batch()290 for doc in r.results:291 yield pickle.loads(doc[self.descriptor_field])292 def iteritems(self):293 """294 Return an iterator over set descriptor key and instance pairs.295 """296 r = self.solr.select('%s:%s %s:* %s:*'297 % (self.set_uuid_field, self.set_uuid,298 self.d_uid_field, self.descriptor_field))299 for doc in r.results:300 d = pickle.loads(doc[self.descriptor_field])301 yield d.uuid(), d302 for _ in range(r.numFound // 10):303 r = r.next_batch()304 for doc in r.results:305 d = pickle.loads(doc[self.descriptor_field])...
test_pi_service.py
Source:test_pi_service.py
...37 return json_port38@pytest.fixture39def fabric():40 fabric = vnc_api.Fabric("fabric-name")41 fabric.set_uuid("fabric-uuid-1")42 return fabric43@pytest.fixture44def physical_router(fabric):45 pr = vnc_api.PhysicalRouter("qfx-1")46 pr.set_uuid("pr-1-uuid")47 pr.add_fabric(fabric)48 return pr49@pytest.fixture50def physical_interface(physical_router):51 pi = vnc_api.PhysicalInterface("xe-0/0/1", physical_router)52 pi.set_uuid("pi-1-uuid")53 pi.parent_uuid = physical_router.uuid54 return pi55@pytest.fixture56def fabric_1():57 fabric_1 = vnc_api.Fabric(name="fabric_1")58 fabric_1.set_uuid("fabric-uuid-1")59 return fabric_160@pytest.fixture61def fabric_2():62 fabric_2 = vnc_api.Fabric(name="fabric_2")63 fabric_2.set_uuid("fabric-uuid-2")64 return fabric_265@pytest.fixture66def physical_router_1(fabric_1):67 pr = vnc_api.PhysicalRouter(name="pr-1-1")68 pr.set_uuid("pr-uuid-1-1")69 pr.add_fabric(fabric_1)70 return pr71@pytest.fixture72def physical_router_2(fabric_1):73 pr = vnc_api.PhysicalRouter(name="pr-1-2")74 pr.set_uuid("pr-uuid-1-2")75 pr.add_fabric(fabric_1)76 return pr77@pytest.fixture78def physical_router_3(fabric_2):79 pr = vnc_api.PhysicalRouter(name="pr-2-1")80 pr.set_uuid("pr-uuid-2-1")81 pr.add_fabric(fabric_2)82 return pr83@pytest.fixture84def physical_router_4():85 pr = vnc_api.PhysicalRouter(name="pr-2-2")86 pr.set_uuid("pr-uuid-2-2")87 return pr88def test_populate_db(89 pi_service,90 vcenter_api_client,91 vnc_api_client,92 database,93 host,94 node,95 port,96 invalid_port,97 physical_interface,98 physical_router,99):100 vcenter_api_client.get_all_hosts.return_value = [host]...
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 yandex-tank 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?
