マルチスレッドはてなキーワードしりとりを改造・その４

前回のコードではmulti_input_reader関数がmain関数のローカル変数である"usedwords"を引数としていたためmainとmulti_input_readerの分離が悪い、と書いていましたが、この問題への対処としてmulti_input_reader関数は引数を取るのをやめ入力の受付のみに特化させました。次に入力する単語がいくつ目になるかはmain関数で表示するようにしました。

前回までのコードでは例えば「はてなダイアリー、、Linux」のように"、"または","で区切って入力をリスト化した際、空白文字列が生じるような場合に、空白文字列をAPIに問い合わせてしまっていましたがfilterを使って入力時に空白をチェックするようにしました。

# coding: utf-8

import shiritori
import threading

WORDS_MAX = 10  # 一度に入力を受け付ける最大単語数


class PrlSupportHFD(shiritori.HatenaFuriganaDict):
    """親クラスの辞書登録処理に排他制御を加える"""
    def __init__(self):
        self.lock = threading.Lock()
        shiritori.HatenaFuriganaDict.__init__(self)

    def __setitem__(self, key, value):
        self.lock.acquire()
        shiritori.HatenaFuriganaDict.__setitem__(self, key, value)
        self.lock.release()


class ParallelAsk(threading.Thread):
    """マルチスレッド関連処理"""
    _furiganadict = PrlSupportHFD()

    def __init__(self, word):
        self.word = word
        threading.Thread.__init__(self)

    def run(self):
        self.result = ParallelAsk._furiganadict[self.word]

    @classmethod
    def get_result(cls, words, wordsmax=WORDS_MAX):
        """問い合わせ結果をwordsの順に
           (ふりがな, 元の単語)のタプルで
           返すジェネレータ
        """
        if len(words) > wordsmax:
            raise WordsInputError(value=wordsmax)

        if len(words) != len(set(words)):
            raise RepeatedWordsError()

        # 子スレッドをセット
        askers = []
        for word in words:
            asker = cls(word)
            askers.append(asker)
            asker.start()

        #各スレッドの結果をFuriganaインスタンス、元単語のタプルで返す
        for (asker, word) in zip(askers, words):
            asker.join()
            furigana = shiritori.Furigana(asker.result, word)

            yield (furigana, word)


def multi_input_reader():
    """入力を受けとり単語のリストを作成する
       ジェネレータ
    """
    while True:
        wordsstr = raw_input(u"次の単語を入力してください。")

        wordsstrU = unicode(wordsstr).replace(u"、", u",")

        wordsU = filter((lambda word: word != u""), wordsstrU.split(u","))

        yield wordsU


class WordsInputError(shiritori.ShiritoriRuleError):
    def __str__(self):
        return (u"入力した単語が多すぎます。一度に入力できるのは"
               + str(self.value)
               + u"語までです。")


class RepeatedWordsError(shiritori.ShiritoriRuleError):
    def __str__(self):
        return u"入力した単語に重複があります。"


def main(initial=u"あ", inputsource=multi_input_reader()):
    usedwords = {}

    print (u"最初のもじは「" + initial + u"」です")

    for wordsU in inputsource:
        try:
            for (furigana, word) in ParallelAsk.get_result(wordsU):
                if word in usedwords:  # 使用済みでないか確認
                    raise shiritori.UniquenessError(word, usedwords[word])

                furigana.exam(initial, word)  # 冒頭、末尾文字確認
                initial = furigana.creatIL()
                print word
                usedwords[word] = len(usedwords) + 1

            print u"--次は" + str(len(usedwords) + 1) + u"語目です。--"

        except shiritori.CrtShiritoriError, e:
            print e
            break
        except shiritori.ShiritoriRuleError, e:
            print e
            print u"--次は" + str(len(usedwords) + 1) + u"語目です。--"
            continue

    print u"終了です"

if __name__ == "__main__":
    main()

さらに、今回はじめてテストを書いてみました。ただテストの書き方以上に「何をテストしたらいいのか」が自分でもよく分かっていなくて正しいテストの書き方かはわかりません(^_^;)
以下テスト

# coding: utf-8

import unittest
import multithread_shiritori
import shiritori


class TestSequence(unittest.TestCase):

    def testHFD(self):
        """辞書テスト"""
        testdict = multithread_shiritori.PrlSupportHFD()
        # 正しいxmlが戻る場合
        self.assertEqual(testdict[u"Linux"], u"りなっくす")
        # はてなキーワードに登録がない場合
        self.assertIsNone(testdict[u"あああ"])
        # xmlにエラーが発生する場合
        self.assertIs(testdict[u"伯爵"], False)

    def testPA(self):
        """子スレッド生成テスト"""
        words = [u"あああ", u"Linux", u"伯爵"]

        for (furigana, word) in multithread_shiritori.ParallelAsk.get_result(words, 10):
            self.assertEqual(furigana, shiritori.Furigana(word))

    def testmain1(self):
        """問題なく６単語が入力されるパターン"""
        wordlist = [
                    [u"アフロ", u"ローマ字"],
                    [u"ジントニック", u"クラス"],
                    [u"スルメ", u"明太子"]
                   ]
        multithread_shiritori.main(u"あ", wordlist)

    def testmain2(self):
        """例外発生パターン"""
        wordlist = [
                    [u"秋葉原", u"ラスベガス"],
                    [u"テスト", u"クラス"],
                    [u"ドレミ", u"カラス"]
                   ]
        multithread_shiritori.main(u"あ", wordlist)

if __name__ == "__main__":
    unittest.main()