tests: Improve stream.c test coverage.

1 files changed, 132 insertions, 1 deletions

diff --git a/unix/coverage.c b/unix/coverage.c
index 7140be41b9..8b46fd0ff3 100644
--- a/unix/coverage.c
+++ b/unix/coverage.c
@@ -1,4 +1,5 @@
 #include <stdio.h>
+#include <string.h>
 
 #include "py/obj.h"
 #include "py/objstr.h"
@@ -8,6 +9,127 @@
 #include "py/builtin.h"
 #include "py/emit.h"
 #include "py/formatfloat.h"
+#include "py/stream.h"
+
+// stream testing object
+typedef struct _mp_obj_streamtest_t {
+    mp_obj_base_t base;
+    uint8_t *buf;
+    size_t len;
+    size_t pos;
+    int error_code;
+} mp_obj_streamtest_t;
+
+STATIC mp_obj_t stest_set_buf(mp_obj_t o_in, mp_obj_t buf_in) {
+    mp_obj_streamtest_t *o = MP_OBJ_TO_PTR(o_in);
+    mp_buffer_info_t bufinfo;
+    mp_get_buffer_raise(buf_in, &bufinfo, MP_BUFFER_READ);
+    o->buf = m_new(uint8_t, bufinfo.len);
+    memcpy(o->buf, bufinfo.buf, bufinfo.len);
+    o->len = bufinfo.len;
+    o->pos = 0;
+    return mp_const_none;
+}
+STATIC MP_DEFINE_CONST_FUN_OBJ_2(stest_set_buf_obj, stest_set_buf);
+
+STATIC mp_obj_t stest_set_error(mp_obj_t o_in, mp_obj_t err_in) {
+    mp_obj_streamtest_t *o = MP_OBJ_TO_PTR(o_in);
+    o->error_code = mp_obj_get_int(err_in);
+    return mp_const_none;
+}
+STATIC MP_DEFINE_CONST_FUN_OBJ_2(stest_set_error_obj, stest_set_error);
+
+STATIC mp_uint_t stest_read(mp_obj_t o_in, void *buf, mp_uint_t size, int *errcode) {
+    mp_obj_streamtest_t *o = MP_OBJ_TO_PTR(o_in);
+    if (o->pos < o->len) {
+        if (size > o->len - o->pos) {
+            size = o->len - o->pos;
+        }
+        memcpy(buf, o->buf + o->pos, size);
+        o->pos += size;
+        return size;
+    } else if (o->error_code == 0) {
+        return 0;
+    } else {
+        *errcode = o->error_code;
+        return MP_STREAM_ERROR;
+    }
+}
+
+STATIC mp_uint_t stest_write(mp_obj_t o_in, const void *buf, mp_uint_t size, int *errcode) {
+    mp_obj_streamtest_t *o = MP_OBJ_TO_PTR(o_in);
+    (void)buf;
+    (void)size;
+    *errcode = o->error_code;
+    return MP_STREAM_ERROR;
+}
+
+STATIC mp_uint_t stest_ioctl(mp_obj_t o_in, mp_uint_t request, uintptr_t arg, int *errcode) {
+    mp_obj_streamtest_t *o = MP_OBJ_TO_PTR(o_in);
+    (void)arg;
+    (void)request;
+    (void)errcode;
+    if (o->error_code != 0) {
+        *errcode = o->error_code;
+        return MP_STREAM_ERROR;
+    }
+    return 0;
+}
+
+STATIC const mp_rom_map_elem_t rawfile_locals_dict_table[] = {
+    { MP_ROM_QSTR(MP_QSTR_set_buf), MP_ROM_PTR(&stest_set_buf_obj) },
+    { MP_ROM_QSTR(MP_QSTR_set_error), MP_ROM_PTR(&stest_set_error_obj) },
+    { MP_ROM_QSTR(MP_QSTR_read), MP_ROM_PTR(&mp_stream_read_obj) },
+    { MP_ROM_QSTR(MP_QSTR_read1), MP_ROM_PTR(&mp_stream_read1_obj) },
+    { MP_ROM_QSTR(MP_QSTR_write), MP_ROM_PTR(&mp_stream_write_obj) },
+    { MP_ROM_QSTR(MP_QSTR_write1), MP_ROM_PTR(&mp_stream_write1_obj) },
+    { MP_ROM_QSTR(MP_QSTR_readinto), MP_ROM_PTR(&mp_stream_readinto_obj) },
+    { MP_ROM_QSTR(MP_QSTR_readline), MP_ROM_PTR(&mp_stream_unbuffered_readline_obj) },
+    { MP_ROM_QSTR(MP_QSTR_ioctl), MP_ROM_PTR(&mp_stream_ioctl_obj) },
+};
+
+STATIC MP_DEFINE_CONST_DICT(rawfile_locals_dict, rawfile_locals_dict_table);
+
+STATIC const mp_stream_p_t fileio_stream_p = {
+    .read = stest_read,
+    .write = stest_write,
+    .ioctl = stest_ioctl,
+};
+
+STATIC const mp_obj_type_t mp_type_stest_fileio = {
+    { &mp_type_type },
+    .protocol = &fileio_stream_p,
+    .locals_dict = (mp_obj_dict_t*)&rawfile_locals_dict,
+};
+
+// stream read returns non-blocking error
+STATIC mp_uint_t stest_read2(mp_obj_t o_in, void *buf, mp_uint_t size, int *errcode) {
+    (void)o_in;
+    (void)buf;
+    (void)size;
+    *errcode = MP_EAGAIN;
+    return MP_STREAM_ERROR;
+}
+
+STATIC const mp_rom_map_elem_t rawfile_locals_dict_table2[] = {
+    { MP_ROM_QSTR(MP_QSTR_read), MP_ROM_PTR(&mp_stream_read_obj) },
+    { MP_ROM_QSTR(MP_QSTR_write), MP_ROM_PTR(&mp_stream_write_obj) },
+};
+
+STATIC MP_DEFINE_CONST_DICT(rawfile_locals_dict2, rawfile_locals_dict_table2);
+
+STATIC const mp_stream_p_t textio_stream_p2 = {
+    .read = stest_read2,
+    .write = NULL,
+    .is_text = true,
+};
+
+STATIC const mp_obj_type_t mp_type_stest_textio2 = {
+    { &mp_type_type },
+    .protocol = &textio_stream_p2,
+    .locals_dict = (mp_obj_dict_t*)&rawfile_locals_dict2,
+};
+
 
 #if defined(MICROPY_UNIX_COVERAGE)
 
@@ -157,8 +279,17 @@ STATIC mp_obj_t extra_coverage(void) {
         mp_printf(&mp_plat_print, "%s\n", buf2);
     }
 
+    mp_obj_streamtest_t *s = m_new_obj(mp_obj_streamtest_t);
+    s->base.type = &mp_type_stest_fileio;
+    s->buf = NULL;
+    s->len = 0;
+    s->pos = 0;
+    s->error_code = 0;
+    mp_obj_streamtest_t *s2 = m_new_obj(mp_obj_streamtest_t);
+    s2->base.type = &mp_type_stest_textio2;
+
     // return a tuple of data for testing on the Python side
-    mp_obj_t items[] = {(mp_obj_t)&str_no_hash_obj, (mp_obj_t)&bytes_no_hash_obj};
+    mp_obj_t items[] = {(mp_obj_t)&str_no_hash_obj, (mp_obj_t)&bytes_no_hash_obj, MP_OBJ_FROM_PTR(s), MP_OBJ_FROM_PTR(s2)};
     return mp_obj_new_tuple(MP_ARRAY_SIZE(items), items);
 }
 MP_DEFINE_CONST_FUN_OBJ_0(extra_coverage_obj, extra_coverage);