| 1 | """ |
|---|
| 2 | Test cases for cddb protocol |
|---|
| 3 | """ |
|---|
| 4 | |
|---|
| 5 | from twisted.internet import protocol |
|---|
| 6 | from twisted.trial import unittest |
|---|
| 7 | import cddbp |
|---|
| 8 | import StringIO |
|---|
| 9 | |
|---|
| 10 | class FakeFile(StringIO.StringIO): |
|---|
| 11 | def close(self): |
|---|
| 12 | pass |
|---|
| 13 | |
|---|
| 14 | class CDDBPClientTester(cddbp.CDDBPClient): |
|---|
| 15 | def connectionMade(self): |
|---|
| 16 | cddbp.CDDBPClient.connectionMade(self) |
|---|
| 17 | self.empty() |
|---|
| 18 | |
|---|
| 19 | def empty(self): |
|---|
| 20 | self.line = "" |
|---|
| 21 | self.err_line = "" |
|---|
| 22 | self.data = "" |
|---|
| 23 | |
|---|
| 24 | def handleStatus(self, code, line): |
|---|
| 25 | if code in cddbp.successful_codes: |
|---|
| 26 | self.line = line |
|---|
| 27 | elif code in cddbp.multi_line_codes: |
|---|
| 28 | self.multi_line = True |
|---|
| 29 | elif code in cddbp.failure_error_codes: |
|---|
| 30 | self.err_line = line |
|---|
| 31 | elif code in [401, 402, 500, cddbp.NO_MATCH_FOUND, |
|---|
| 32 | cddbp.DB_ENTRY_CORRUPTED, cddbp.INVALID_DATA, |
|---|
| 33 | cddbp.SAME_PROTO_LEVEL, cddbp.QUIT_ERROR_CLOSING]: |
|---|
| 34 | self.err_line = line |
|---|
| 35 | |
|---|
| 36 | def parseResponse(self, response): |
|---|
| 37 | pass |
|---|
| 38 | |
|---|
| 39 | class CDDBPClientTestCase(unittest.TestCase): |
|---|
| 40 | tester = f = None |
|---|
| 41 | full_discid = ["f2123610", "16", "150", "29977", "46577", "68970", |
|---|
| 42 | "85297", "104922", "131622", "150317", "157300", |
|---|
| 43 | "181212", "208612", "231910", "253045", "273352", |
|---|
| 44 | "295987", "326627", "4664"] |
|---|
| 45 | cd_data = full_discid[1:] |
|---|
| 46 | discid = full_discid[0] |
|---|
| 47 | full_discid = " ".join(full_discid) |
|---|
| 48 | cd_data = " ".join(cd_data) |
|---|
| 49 | |
|---|
| 50 | def setUp(self): |
|---|
| 51 | self.tester = CDDBPClientTester() |
|---|
| 52 | self.f = FakeFile() |
|---|
| 53 | self.tester.makeConnection(protocol.FileWrapper(self.f)) |
|---|
| 54 | |
|---|
| 55 | def checkOK(self, line): |
|---|
| 56 | self.tester.lineReceived(line) |
|---|
| 57 | self.assertEqual(self.tester.line, line) |
|---|
| 58 | self.failIf(self.tester.err_line) |
|---|
| 59 | self.failIf(self.tester.multi_line) |
|---|
| 60 | |
|---|
| 61 | def checkErr(self, line): |
|---|
| 62 | self.tester.lineReceived(line) |
|---|
| 63 | self.assertEqual(self.tester.err_line, line) |
|---|
| 64 | self.failIf(self.tester.line) |
|---|
| 65 | self.failIf(self.tester.multi_line) |
|---|
| 66 | |
|---|
| 67 | def checkMultiLine(self, line): |
|---|
| 68 | self.tester.lineReceived(line) |
|---|
| 69 | self.assert_(self.tester.multi_line) |
|---|
| 70 | |
|---|
| 71 | def isBufferNotEmpty(self, line): |
|---|
| 72 | self.tester.lineReceived(line) |
|---|
| 73 | self.assertEqual(self.tester.buffer, [line]) |
|---|
| 74 | |
|---|
| 75 | def isBufferEmpty(self, line): |
|---|
| 76 | self.tester.lineReceived(line) |
|---|
| 77 | self.failIf(self.tester.buffer) |
|---|
| 78 | self.failIf(self.tester.multi_line) |
|---|
| 79 | |
|---|
| 80 | def checkBuffer(self, line): |
|---|
| 81 | self.isBufferNotEmpty(line) |
|---|
| 82 | line = cddbp.TERMINATING_MARKER |
|---|
| 83 | self.isBufferEmpty(line) |
|---|
| 84 | |
|---|
| 85 | def testLoginOk(self): |
|---|
| 86 | import time |
|---|
| 87 | code = cddbp.OK_READ_ONLY |
|---|
| 88 | line = "%s foohost CDDBP server v1.0 ready %s" % (code, time.asctime()) |
|---|
| 89 | self.tester.lineReceived(line) |
|---|
| 90 | self.checkOK(line) |
|---|
| 91 | |
|---|
| 92 | def testLoginNotAllowed(self): |
|---|
| 93 | line = "%s no connections allowed" % cddbp.CONNECT_PERMISSION_DENIED |
|---|
| 94 | self.tester.lineReceived(line) |
|---|
| 95 | self.checkErr(line) |
|---|
| 96 | |
|---|
| 97 | def testHelloHandshakeSuccessful(self): |
|---|
| 98 | username, host, app, version = "foo", "localhost", "bar", "0.1" |
|---|
| 99 | self.tester.hello(username, host, app, version) |
|---|
| 100 | line = "%s hello and welcome %s@%s running %s %s" %\ |
|---|
| 101 | (cddbp.OK, |
|---|
| 102 | username, |
|---|
| 103 | host, |
|---|
| 104 | app, version) |
|---|
| 105 | self.checkOK(line) |
|---|
| 106 | |
|---|
| 107 | def testHelloHandshakeFailed(self): |
|---|
| 108 | username, host, app, version = "foo", "localhost", "bar", "0.1" |
|---|
| 109 | self.tester.hello(username, host, app, version) |
|---|
| 110 | line = "%s handshake failed" % cddbp.HANDSHAKE_FAILED |
|---|
| 111 | self.checkErr(line) |
|---|
| 112 | |
|---|
| 113 | def testLscat(self): |
|---|
| 114 | self.tester.lscat() |
|---|
| 115 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 116 | line = "%s Okay category list follows (until terminating marker)" % code |
|---|
| 117 | self.checkMultiLine(line) |
|---|
| 118 | line = "test" |
|---|
| 119 | self.isBufferNotEmpty(line) |
|---|
| 120 | line = cddbp.TERMINATING_MARKER |
|---|
| 121 | self.isBufferEmpty(line) |
|---|
| 122 | |
|---|
| 123 | def testQueryExactMatch(self): |
|---|
| 124 | self.tester.query(self.full_discid) |
|---|
| 125 | code = cddbp.EXACT_MATCH |
|---|
| 126 | line = "%s misc f2123610 Common / Like Water For Chocolate" % code |
|---|
| 127 | self.checkOK(line) |
|---|
| 128 | |
|---|
| 129 | def testQueryMultipleInexactMatches(self): |
|---|
| 130 | self.tester.query(self.full_discid) |
|---|
| 131 | line = "%s close matches found" % cddbp.MULTIPLE_INEXACT_MATCHES |
|---|
| 132 | self.checkMultiLine(line) |
|---|
| 133 | line = "misc f2123610 Common / Like Water For Chocolate" |
|---|
| 134 | self.checkBuffer(line) |
|---|
| 135 | |
|---|
| 136 | testQueryMultipleExactMatches = testQueryMultipleInexactMatches |
|---|
| 137 | |
|---|
| 138 | def testQueryMatchNotFound(self): |
|---|
| 139 | self.tester.query(self.full_discid) |
|---|
| 140 | line = "%s no match found" % cddbp.NO_MATCH_FOUND |
|---|
| 141 | self.checkErr(line) |
|---|
| 142 | |
|---|
| 143 | def testQueryDBEntryCorrupted(self): |
|---|
| 144 | self.tester.query(self.full_discid) |
|---|
| 145 | line = "%s database entry is corrupt" % cddbp.DB_ENTRY_CORRUPTED |
|---|
| 146 | self.checkErr(line) |
|---|
| 147 | |
|---|
| 148 | def testReadEntryFollows(self): |
|---|
| 149 | self.tester.read("misc", self.discid) |
|---|
| 150 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 151 | line = "%s OK, CDDB database entry follows" % code |
|---|
| 152 | self.checkMultiLine(line) |
|---|
| 153 | line = "test" |
|---|
| 154 | self.checkBuffer(line) |
|---|
| 155 | |
|---|
| 156 | def testReadMatchNotFound(self): |
|---|
| 157 | self.tester.read("misc", self.discid) |
|---|
| 158 | line = "%s specified CDDB entry not found." % cddbp.NOT_AVAILABLE |
|---|
| 159 | self.checkErr(line) |
|---|
| 160 | |
|---|
| 161 | def testReadServerError(self): |
|---|
| 162 | self.tester.read("misc", self.discid) |
|---|
| 163 | line = "%s server error." % cddbp.SERVER_ERROR |
|---|
| 164 | self.checkErr(line) |
|---|
| 165 | |
|---|
| 166 | def testReadDBEntryCorrupted(self): |
|---|
| 167 | self.tester.read("misc", self.discid) |
|---|
| 168 | line = "%s server error." % cddbp.DB_ENTRY_CORRUPTED |
|---|
| 169 | self.checkErr(line) |
|---|
| 170 | |
|---|
| 171 | def testDiscidComputed(self): |
|---|
| 172 | self.tester.discid(self.cd_data) |
|---|
| 173 | line = "%s Discid is %s" % (cddbp.OK, self.discid) |
|---|
| 174 | self.checkOK(line) |
|---|
| 175 | |
|---|
| 176 | def testHelp(self): |
|---|
| 177 | self.tester.help() |
|---|
| 178 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 179 | line = "%s OK, help information follows" % code |
|---|
| 180 | self.checkMultiLine(line) |
|---|
| 181 | line = "test" |
|---|
| 182 | self.checkBuffer(line) |
|---|
| 183 | |
|---|
| 184 | def testHelpNotAvailable(self): |
|---|
| 185 | self.tester.help("quit") |
|---|
| 186 | code = cddbp.NOT_AVAILABLE |
|---|
| 187 | line = "%s No help information available" % code |
|---|
| 188 | self.checkErr(line) |
|---|
| 189 | |
|---|
| 190 | def testMotd(self): |
|---|
| 191 | self.tester.motd() |
|---|
| 192 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 193 | line = "%s OK, Last modified: 01/02/03 01:02:03 MOTD follows" % code |
|---|
| 194 | self.checkMultiLine(line) |
|---|
| 195 | line = "test" |
|---|
| 196 | self.checkBuffer(line) |
|---|
| 197 | |
|---|
| 198 | def testMotdNotAvailable(self): |
|---|
| 199 | self.tester.motd() |
|---|
| 200 | code = cddbp.NOT_AVAILABLE |
|---|
| 201 | line = "%s No message of the day available" % code |
|---|
| 202 | self.checkErr(line) |
|---|
| 203 | |
|---|
| 204 | def testProtoGet(self): |
|---|
| 205 | self.tester.proto() |
|---|
| 206 | code = cddbp.CURRENT_PROTO_LEVEL |
|---|
| 207 | line = "%s CDDB protocol level: current 1, supported 6" % code |
|---|
| 208 | self.checkOK(line) |
|---|
| 209 | |
|---|
| 210 | def testProtoCorrectSet(self): |
|---|
| 211 | level = "6" |
|---|
| 212 | self.tester.proto(level) |
|---|
| 213 | code = cddbp.OK_PROTO_LEVEL_CHANGED |
|---|
| 214 | line = "%s OK, protocol version now: %s" % (code, level) |
|---|
| 215 | self.checkOK(line) |
|---|
| 216 | |
|---|
| 217 | def testProtoIllegalSet(self): |
|---|
| 218 | self.tester.proto("9") |
|---|
| 219 | code = cddbp.INVALID_DATA |
|---|
| 220 | line = "%s Illegal protocol level." % code |
|---|
| 221 | self.checkErr(line) |
|---|
| 222 | |
|---|
| 223 | def testProtoSameLevel(self): |
|---|
| 224 | level = "1" |
|---|
| 225 | self.tester.proto(level) |
|---|
| 226 | code = cddbp.SAME_PROTO_LEVEL |
|---|
| 227 | line = "%s OK, protocol version now: %s" % (code, level) |
|---|
| 228 | self.checkErr(line) |
|---|
| 229 | |
|---|
| 230 | def testQuitOkClosing(self): |
|---|
| 231 | self.tester.quit() |
|---|
| 232 | code = cddbp.QUIT_OK_CLOSING |
|---|
| 233 | line = "%s foo.bar.com Closing connection. Goodbye." % code |
|---|
| 234 | self.checkOK(line) |
|---|
| 235 | |
|---|
| 236 | def testQuitErrClosing(self): |
|---|
| 237 | self.tester.quit() |
|---|
| 238 | code = cddbp.QUIT_ERROR_CLOSING |
|---|
| 239 | line = "%s foo.bar.com error, closing connection." % code |
|---|
| 240 | self.checkErr(line) |
|---|
| 241 | |
|---|
| 242 | def testSites(self): |
|---|
| 243 | self.tester.sites() |
|---|
| 244 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 245 | line = "%s OK, site information follows" % code |
|---|
| 246 | self.checkMultiLine(line) |
|---|
| 247 | line = "test" |
|---|
| 248 | self.checkBuffer(line) |
|---|
| 249 | |
|---|
| 250 | def testSitesNotAvailable(self): |
|---|
| 251 | self.tester.sites() |
|---|
| 252 | code = cddbp.NOT_AVAILABLE |
|---|
| 253 | line = "%s No site information available" % code |
|---|
| 254 | self.checkErr(line) |
|---|
| 255 | |
|---|
| 256 | def testStat(self): |
|---|
| 257 | self.tester.stat() |
|---|
| 258 | code = cddbp.OK_FURTHER_DATA_FOLLOWS |
|---|
| 259 | line = "%s OK, status information follows" % code |
|---|
| 260 | self.checkMultiLine(line) |
|---|
| 261 | line = "test" |
|---|
| 262 | self.checkBuffer(line) |
|---|
| 263 | |
|---|
| 264 | def testVer(self): |
|---|
| 265 | self.tester.ver() |
|---|
| 266 | code = cddbp.OK |
|---|
| 267 | line = "%s fooserver 1.0 Copyright (c) 2004 Spider Man" % code |
|---|
| 268 | self.checkOK(line) |
|---|
| 269 | |
|---|
| 270 | def testVerMultiline(self): |
|---|
| 271 | self.tester.ver() |
|---|
| 272 | code = cddbp.OK_VERSION_INFO_FOLLOWS |
|---|
| 273 | line = "%s OK, Version information follows" % code |
|---|
| 274 | self.checkMultiLine(line) |
|---|
| 275 | line = "test" |
|---|
| 276 | self.checkBuffer(line) |
|---|
| 277 | |
|---|
| 278 | #def testUnlinkOk(self): |
|---|
| 279 | #self.tester.unlink("misc", self.discid) |
|---|
| 280 | #code = cddbp.OK |
|---|
| 281 | #line = "%s OK, file has been deleted." % code |
|---|
| 282 | #self.checkOK(line) |
|---|
| 283 | |
|---|
| 284 | #def testUnlinkDenied(self): |
|---|
| 285 | #self.tester.unlink("misc", self.discid) |
|---|
| 286 | #code = cddbp.UNLINK_PERMISSION_DENIED |
|---|
| 287 | #line = "%s Permission denied." % code |
|---|
| 288 | #self.checkErr(line) |
|---|
| 289 | |
|---|
| 290 | #def testUnlinkFileAccessFailed(self): |
|---|
| 291 | #self.tester.unlink("misc", self.discid) |
|---|
| 292 | #code = cddbp.FILE_ACCESS_FAILED |
|---|
| 293 | #line = "%s File access failed." % code |
|---|
| 294 | #self.checkErr(line) |
|---|
| 295 | |
|---|
| 296 | #def testUnlinkInvalidCategory(self): |
|---|
| 297 | #self.tester.unlink("misc", self.discid) |
|---|
| 298 | #code = cddbp.INVALID_DATA |
|---|
| 299 | #line = "%s Invalid category: %s" % (code, "misc") |
|---|
| 300 | #self.checkErr(line) |
|---|
| 301 | |
|---|
| 302 | def test500Codes(self): |
|---|
| 303 | self.tester.sendLine("foo bar") |
|---|
| 304 | # since cddbp uses the same code for 3 differents things, testing |
|---|
| 305 | # one or another it's the same |
|---|
| 306 | line = "%s blah blah" % cddbp.UNKOWN_COMMAND |
|---|
| 307 | self.checkErr(line) |
|---|
| 308 | |
|---|
| 309 | class CDDBPServerTester(cddbp.CDDBPServer): |
|---|
| 310 | pass |
|---|
| 311 | |
|---|
| 312 | class CDDBPServerTestCase(unittest.TestCase): |
|---|
| 313 | def testClientLogged(self): |
|---|
| 314 | pass |
|---|