(self, r, slowlog)
| 1182 | |
| 1183 | @pytest.mark.onlynoncluster |
| 1184 | def test_slowlog_get(self, r, slowlog): |
| 1185 | assert r.slowlog_reset() |
| 1186 | unicode_string = chr(3456) + "abcd" + chr(3421) |
| 1187 | r.get(unicode_string) |
| 1188 | slowlog = r.slowlog_get() |
| 1189 | assert isinstance(slowlog, list) |
| 1190 | commands = [log["command"] for log in slowlog] |
| 1191 | |
| 1192 | get_command = b" ".join((b"GET", unicode_string.encode("utf-8"))) |
| 1193 | assert get_command in commands |
| 1194 | assert b"SLOWLOG RESET" in commands |
| 1195 | # the order should be ['GET <uni string>', 'SLOWLOG RESET'], |
| 1196 | # but if other clients are executing commands at the same time, there |
| 1197 | # could be commands, before, between, or after, so just check that |
| 1198 | # the two we care about are in the appropriate order. |
| 1199 | assert commands.index(get_command) < commands.index(b"SLOWLOG RESET") |
| 1200 | |
| 1201 | # make sure other attributes are typed correctly |
| 1202 | assert isinstance(slowlog[0]["start_time"], int) |
| 1203 | assert isinstance(slowlog[0]["duration"], int) |
| 1204 | assert isinstance(slowlog[0]["client_address"], bytes) |
| 1205 | assert isinstance(slowlog[0]["client_name"], bytes) |
| 1206 | |
| 1207 | # Mock result if we didn't get slowlog complexity info. |
| 1208 | if "complexity" not in slowlog[0]: |
| 1209 | # monkey patch parse_response() |
| 1210 | COMPLEXITY_STATEMENT = "Complexity info: N:4712,M:3788" |
| 1211 | old_parse_response = r.parse_response |
| 1212 | |
| 1213 | def parse_response(connection, command_name, **options): |
| 1214 | if command_name != "SLOWLOG GET": |
| 1215 | return old_parse_response(connection, command_name, **options) |
| 1216 | responses = connection.read_response() |
| 1217 | for response in responses: |
| 1218 | # Complexity info stored as fourth item in list |
| 1219 | response.insert(3, COMPLEXITY_STATEMENT) |
| 1220 | return r.response_callbacks[command_name](responses, **options) |
| 1221 | |
| 1222 | r.parse_response = parse_response |
| 1223 | |
| 1224 | # test |
| 1225 | slowlog = r.slowlog_get() |
| 1226 | assert isinstance(slowlog, list) |
| 1227 | commands = [log["command"] for log in slowlog] |
| 1228 | assert get_command in commands |
| 1229 | idx = commands.index(get_command) |
| 1230 | assert slowlog[idx]["complexity"] == COMPLEXITY_STATEMENT |
| 1231 | |
| 1232 | # tear down monkeypatch |
| 1233 | r.parse_response = old_parse_response |
| 1234 | |
| 1235 | @pytest.mark.onlynoncluster |
| 1236 | def test_slowlog_get_limit(self, r, slowlog): |
nothing calls this directly
no test coverage detected