1

最近、サーバーをテストするクライアント テスト プログラムを作成しました。テストでは、orm を使用してデータベース内の既存のオブジェクト (行) を削除します。次に、一連のコマンドをサーバーに送信して、データを再作成します。しかし、テスト プログラムで「削除された」オブジェクトをクエリしようとすると、別の「名前」のセッションでクエリを実行しない限り、クエリはオブジェクトが見つからないことを返します。セッションが SqlAlchemy でどのように機能するかについて懸念がありますが、これがバグであるかどうかはわかりません。次に、この問題をさらに説明します。

engine  = create_engine("mysql:....")
Session = sessionmaker(bind=engine)
session = Session()

for registration in session.query(Registration).filter_by(license_id = id).all():
    session.delete(registration)
    session.commit()

.... データベースの登録テーブルにいくつかの行を作成するコマンドをサーバーに送信するコード ....

x = session.query(Registration).filter_by(license_id = id).all()
print x

x は空のリストになります。

2回目のクエリの前に次を追加しようとすると、すべて同じ結果が返されます

session.close()
session = Session()

また

session.expunge_all()

また

session.exipre_all()

以下を実行するまで、上記のいずれも機能しません

new_session = Session()
x = new_session.query(Registration).filter_by(license_id = id).all()

今度は、x はデータベースに格納されているものを返します。

機能するためにセッション変数名を変更する必要がある理由がちょっとわかりません。session.expire、session.expunge などのセッションに関連するすべてのドキュメント、またはセッションの別のインスタンスを取得すること (同じ名前で別のインスタンスである、確認済み) が機能しません。

誰かが助けてくれれば感謝します。ありがとう。

完全なコード

import socket
import blowfish
import base64
import string
import sys
import logging
import time
from datetime import date, timedelta

from sqlalchemy import create_engine, and_
from sqlalchemy import Column, Integer, String, DateTime, Date, Boolean, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.orm.session import Session

from lrp_modal import User, Product, License, Registration

log = logging.getLogger('')
log.setLevel(logging.INFO)
log.addHandler(logging.StreamHandler())

def test(token, passwd, xmsg, expect_ok):
    serverHost = 'localhost'
    serverPort = 9120

    sockobj = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sockobj.connect((serverHost, serverPort))
    data = sockobj.recv(1024)
    log.debug('Client received: %s', repr(data))

    log.debug("Send: %s",  xmsg)
    sockobj.send(xmsg)

    x = ''
    while 1:
        data = sockobj.recv(1024)
        if not data:
            break
        x += data
    x = x.strip()
    log.debug("Receive: %s", x)
    if expect_ok and not x.startswith('<lrp'):
        print "Test fail"
        sys.exit(1)
    if not expect_ok and x.startswith('<lrp'):
        print "Test fail"
        sys.exit(1)

master   = 'X01X-0000-0000-'
liclist  = {
    'product60_tri': ['0001'],
    'product60_pro': ['0002'],
    'product50_ent': ['0003', '0004', '0005'],
    'product60_ent': ['0006', '0007', '0008'],
    }

machines = ['FFBCC89220EF', '00112233']

# Setup the licenses testing data
engine  = create_engine('mysql://root@127.0.0.1/padsystem')
Session = sessionmaker(bind=engine)
session = Session()

for tag, idlist in liclist.items():
    for id in idlist:
        product = session.query(Product).filter_by(product_tag = tag).one()
        try:
            license = session.query(License).filter_by(license_id = master + id).one()
        except NoResultFound:
            continue

        license.product_id = product.product_id
        license.act_rid  = ''
        license.act_type = ''
        license.act_hint = ''
        license.group_id = ''
        if id == '0003':
            license.lic_expire = date(2011, 1, 1)
            if id in ('0001', '0006'):
                license.lic_expire = date.today() + timedelta(10)
        if id in ('0004', '0005', '0007', '0008'):
            license.group_id = 'X'
        session.commit()

        for registration in session.query(Registration).filter_by(license_id = license.license_id).all():
            session.delete(registration)
        session.commit()

user    = session.query(User).get(1)

# Test to register a machine
log.info('Test register machine...')
data = """<lrp service="register">
    <registration>
      <product>Product50</product>
      <license>%s</license>
      <machine>%s</machine>
    </registration>
  </lrp>
  """
for id in ('0004', '0006', '0007'):
    lic_id = master + id
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[0]), True)
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[1]), True)

for id in ('0001', '0003'):
    lic_id = master + id
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[0]), False)

# Test to register a machine until register count is full.
lic_id = master + '0002'
for i in range(5):
    log.info("Register count test %s ", i)
    x = data % (lic_id, machines[1] + str(i))
    test('LICENSEE %s\n' % user.user_name, user.password, x, True)

log.info("Register exceed max...")
x = data % (lic_id, machines[0])
test('LICENSEE %s\n' % user.user_name, user.password, x, False)

# Test register to group of licenses
lic_id = master + '0008'
for i in range(8):
    log.info("Register group count test: %s ", i)
    x = data % (lic_id, machines[1] + str(i))
    test('LICENSEE %s\n' % user.user_name, user.password, x, True)

log.info("Register exceed group max...")
x = data % (lic_id, machines[0])
test('LICENSEE %s\n' % user.user_name, user.password, x, False)

# Test to unregister
data = """<lrp service="unregister">
    <registration id="%s">
      <product>Product50</product>
      <license>%s</license>
      <machine>%s</machine>
    </registration>
  </lrp>
  """

# session.expunge_all()  not work
# session.expire_all()   not work
# session.close(); del(session); session = Session()  not work
new_session = Session()  # This is needed, cannot use session!!
lic_id = master + '0004'
# Use original session instance will get null list.
for r in new_session.query(Registration).filter_by(license_id = lic_id).all():
    rid = r.rid
    mid = r.machine_id
    log.info("Unregister, wrong license...")
    x = data % (rid, master + '0005', mid)
    test('REGISTER %s\n' % rid, mid, x, False)
    log.info("Unregister...")
    x = data % (rid, lic_id, mid) 
    test('REGISTER %s\n' % rid, mid, x, True)
4

0 に答える 0